Android如何使用Flow封装一个FlowBus工具类_Android

android中使用flow封装一个flowbus工具类

做过android的同学应该都使用过evenutbus、rxbus、livedatabus、livedata等，这些第三方不仅要导入依赖包，而且还要注册和取消注册，使用起来非常麻烦，稍不注意就导致内存泄漏，自从接触了flow、sharedflow之后感觉使用起来方便多了，于是产生了一个封装通用事件工具类的想法,直接上代码.

1.flowbus:

/**
 * @auth: njb
 * @date: 2024/7/18 10:17
 * @desc: 基于flow封装的flowbus
 */
object flowbus {
    private const val tag = "flowbus"
    private val busmap = mutablemapof<string, floweventbus<*>>()
    private val busstickmap = mutablemapof<string, flowstickeventbus<*>>()
    @synchronized
    fun <t> with(key: string): floweventbus<t> {
        var floweventbus = busmap[key]
        if (floweventbus == null) {
            floweventbus = floweventbus<t>(key)
            busmap[key] = floweventbus
        }
        return floweventbus as floweventbus<t>
    }
    @synchronized
    fun <t> withstick(key: string): flowstickeventbus<t> {
        var stickeventbus = busstickmap[key]
        if (stickeventbus == null) {
            stickeventbus = flowstickeventbus<t>(key)
            busstickmap[key] = stickeventbus
        }
        return stickeventbus as flowstickeventbus<t>
    }
    open class floweventbus<t>(private val key: string) : defaultlifecycleobserver {
        //私有对象用于发送消息
        private val _events: mutablesharedflow<t> by lazy {
            obtainevent()
        }
        //暴露的公有对象用于接收消息
        private val events = _events.assharedflow()
        open fun obtainevent(): mutablesharedflow<t> =
            mutablesharedflow(0, 1, bufferoverflow.drop_oldest)
        //在主线程中接收数据
        fun register(lifecycleowner: lifecycleowner,action: (t: t) -> unit){
            lifecycleowner.lifecyclescope.launch {
                events.collect {
                    try {
                        action(it)
                    }catch (e:exception){
                        e.printstacktrace()
                        log.e(tag, "flowbus - error:$e")
                    }
                }
            }
        }
        //在协程中接收数据
        fun register(scope: coroutinescope,action: (t: t) -> unit){
            scope.launch {
                events.collect{
                    try {
                       action(it)
                    }catch (e:exception){
                        e.printstacktrace()
                        log.e(tag, "flowbus - error:$e")
                    }
                }
            }
        }
        //在协程中发送数据
        suspend fun post(event: t){
            _events.emit(event)
        }
        //在主线程中发送数据
        fun post(scope: coroutinescope,event: t){
            scope.launch {
                _events.emit(event)
            }
        }
        override fun ondestroy(owner: lifecycleowner) {
            super.ondestroy(owner)
            log.w(tag, "flowbus ==== 自动ondestroy")
            val subscriptcount = _events.subscriptioncount.value
            if (subscriptcount <= 0)
                busmap.remove(key)
        }
        // 手动调用的销毁方法，用于service、广播等
        fun destroy() {
            log.w(tag, "flowbus ==== 手动销毁")
            val subscriptioncount = _events.subscriptioncount.value
            if (subscriptioncount <= 0) {
                busmap.remove(key)
            }
        }
    }
    class flowstickeventbus<t>(key: string) : floweventbus<t>(key) {
        override fun obtainevent(): mutablesharedflow<t> =
            mutablesharedflow(1, 1, bufferoverflow.drop_oldest)
    }
}

2.在activity中的使用：

2.1传递参数给主界面activity:

/**
 * @auth: njb
 * @date: 2024/9/10 23:49
 * @desc: 描述
 */
class testactivity :appcompatactivity(){
    private val textview:textview by lazy { findviewbyid(r.id.tv_test) }
    override fun oncreate(savedinstancestate: bundle?) {
        super.oncreate(savedinstancestate)
        setcontentview(r.layout.activity_test)
        initflowbus()
    }
    private fun initflowbus() {
        val messageevent = messageevent()
        messageevent.message = "stop"
        messageevent.state = false
        textview.setonclicklistener {
            lifecyclescope.launch {
                flowbus.with<messageevent>("test").post(this, messageevent)
                finish()
            }
        }
    }
}

2.2 mainactivity接收：

/**
 * 初始化
 */
private fun initview() {
    binding.rvwallpaper.apply {
        layoutmanager = gridlayoutmanager(this@mainactivity, 2)
        adapter = wallpaperadapter
    }
    binding.btngetwallpaper.setonclicklistener {
        lifecyclescope.launch {
            mainviewmodel.mainintentchannel.send(mainintent.getwallpaper)
        }
        val intent = intent(this@mainactivity,testactivity::class.java)
        startactivity(intent)
    }
    flowbus.with<messageevent>("test").register(this@mainactivity) {
        logutils.d(tag,it.tostring())
        if(it.message == "stop"){
            logutils.d(tag,"===接收到的消息为==="+it.message)
        }
    }
    flowbus.with<messageevent>("minefragment").register(this@mainactivity) {
        logutils.d(tag,it.tostring())
        if(it.message == "onmine"){
            logutils.d(tag,"===接收到的消息为1111==="+it.message)
        }
    }
}

3.在fragment中的使用：

3.1 发送数据

package com.cloud.flowbusdemo.fragment
import android.os.bundle
import android.util.log
import android.view.layoutinflater
import android.view.view
import android.view.viewgroup
import androidx.fragment.app.fragment
import androidx.lifecycle.lifecyclescope
import com.cloud.flowbusdemo.databinding.fragmentminebinding
import com.cloud.flowbusdemo.flow.flowbus
import com.cloud.flowbusdemo.model.messageevent
import kotlinx.coroutines.launch
private const val arg_param_name = "name"
private const val arg_param_age = "age"
/**
 * @auth: njb
 * @date: 2024/9/17 19:43
 * @desc: 描述
 */
class minefragment :fragment(){
    private lateinit var binding: fragmentminebinding
    private val tag = "minefragment"
    private var name: string? = null
    private var age: int? = null
    override fun oncreate(savedinstancestate: bundle?) {
        super.oncreate(savedinstancestate)
        arguments?.let {
            name = it.getstring(arg_param_name)
            age = it.getint(arg_param_age)
        }
        log.i(tag, "mainfragment 传递到 minefragment 的参数为 name = $name , age = $age")
        log.d(tag, "姓名:" + name + "年龄:" + age)
    }
    override fun oncreateview(
        inflater: layoutinflater,
        container: viewgroup?,
        savedinstancestate: bundle?
    ): view {
        binding = fragmentminebinding.inflate(layoutinflater)
        initview()
        return binding.root
    }
    private fun initview() {
        val messageevent = messageevent()
        messageevent.message = "onmine"
        messageevent.state = false
        binding.let {
            it.tvtitle.text = name
            it.tvage.text  = age.tostring()
            it.tvtitle.setonclicklistener {
                lifecyclescope.launch {
                    flowbus.with<messageevent>("minefragment").post(this, messageevent)
                }
            }
        }
    }
}

3.2 接收数据：

private fun initview() {
    binding.rvwallpaper.apply {
        layoutmanager = gridlayoutmanager(this@mainactivity, 2)
        adapter = wallpaperadapter
    }
    binding.btngetwallpaper.setonclicklistener {
        lifecyclescope.launch {
            mainviewmodel.mainintentchannel.send(mainintent.getwallpaper)
        }
        val intent = intent(this@mainactivity,testactivity::class.java)
        startactivity(intent)
    }
    flowbus.with<messageevent>("test").register(this@mainactivity) {
        logutils.d(tag,it.tostring())
        if(it.message == "stop"){
            logutils.d(tag,"===接收到的消息为==="+it.message)
        }
    }
    flowbus.with<messageevent>("minefragment").register(this@mainactivity) {
        logutils.d(tag,it.tostring())
        if(it.message == "onmine"){
            logutils.d(tag,"===接收到的消息为1111==="+it.message)
        }
    }
}

4.在service中的使用：

4.1发送数据：

private fun initservice() {
    val intent = intent(this@mainactivity, flowbustestservice::class.java)
    intent.putextra("sockurl","")
    startservice(intent)
}

4.2接收数据：

/**
 * @auth: njb
 * @date: 2024/9/22 23:32
 * @desc: 描述
 */
class flowbustestservice:service() {
    private var sock5url:string ?= null
    private val tag = "flowbustestservice"
    override fun onbind(intent: intent?): ibinder? {
        return null
    }
    override fun onstartcommand(intent: intent?, flags: int, startid: int): int {
        intent?.let {
            this.sock5url = intent.getstringextra("sockurl")
            logutils.d(tag,"====收到的ip为==="+this.sock5url)
        }
        return if (intent?.action == constants.action_disconnect) {
            disconnect()
            start_not_sticky
        } else {
            connect()
            start_sticky
        }
    }
    private fun connect() {
    }
    private fun disconnect() {
    }
}

5.在websock中的使用：

5.1发送数据：

private fun connectwebsocket() {
    logutils.e(tag, "===connecturl===$currentwebsocketurl")
    try {
        if (mwebsocketmanager == null) {
            return
        }
        mwebsocketmanager?.addlistener(object : socketlistener {
            override fun onconnected() {
                logutils.e(tag, "===连接成功====")
                val messageevent = messageevent()
                messageevent.message = "socket连接成功"
                floatwindowmanager.log("socket连接成功")
                coroutinescope(dispatchers.main).launch{
                    flowbus.with<messageevent>("onconnected").post(this,messageevent)
                }
            }
            override fun onconnectfailed(throwable: throwable) {
                logutils.e(tag, "===连接失败====")
                val messageevent = messageevent()
                messageevent.message = "socket连接失败:$currentwebsocketurl"
                floatwindowmanager.log("socket连接失败")
            }
            override fun ondisconnect() {
                logutils.e(tag, "===断开连接====")
                val messageevent = messageevent()
                messageevent.message = "socket断开连接"
                floatwindowmanager.log("socket断开连接")
            }
            override fun onsenddataerror(errorresponse: errorresponse) {
                logutils.e(tag + "===发送数据失败====" + errorresponse.description)
                val messageevent = messageevent()
                messageevent.message = "发送数据失败--->" + errorresponse.description
                floatwindowmanager.log("发送数据失败")
            }
            override fun <t> onmessage(msg: string, t: t) {
                logutils.e(tag,"===接收到消息 string===$msg")
                val messageevent = messageevent()
                messageevent.message = msg
                floatwindowmanager.log("===接收到消息===$msg")
                taskmanager?.onhandlemsg(msg)
            }
            override fun <t> onmessage(bytes: bytebuffer, t: t) {
                logutils.e(tag, "===接收到消息bytebuffer===="+gsonutils.tojson(bytes))
                val rbuffer = bytebuffer.allocate(1024)
                val charset = charset.forname("utf-8")
                try {
                    val receivetext =
                        charset.newdecoder().decode(rbuffer.asreadonlybuffer()).tostring()
                    logutils.e(tag, "===接收到消息bytebuffer====$receivetext")
                    val messageevent = messageevent()
                    messageevent.message = receivetext
                   // floatwindowmanager.log("===收到消息 byte===$receivetext")
                } catch (e: charactercodingexception) {
                    throw runtimeexception(e)
                }
            }
            override fun onping(pingdata: framedata) {
                logutils.e(tag, "===心跳onping===$pingdata")
            }
            override fun onpong(framedata: framedata) {
                logutils.e(tag, "===心跳onpong===$framedata")
                val messageevent = messageevent()
                messageevent.message = format.format(date()) + "  | 心跳onpong->"
                floatwindowmanager.log("===心跳onpong===${format.format(date())}${"->"}$currentwebsocketurl")
            }
        })
        mwebsocketmanager?.start()
    } catch (e: exception) {
        e.printstacktrace()
    }
}

5.2接收数据：

private fun initflowbus() {
    flowbus.with<messageevent>("onconnected").register(this@mainactivity) {
        logutils.d(tag, "收到消息为：$it")
    }
    flowbus.with<messageevent>("onstartvpn").register(this@mainactivity) {
        logutils.d(tag, "收到vpn消息为：$it")
        coroutinescope(dispatchers.main).launch {
            if (it.message == "start" && it.state && constants.switch_ip) {
                this@mainactivity.sockurl = it.sockurl
                logutils.d(tag, "收到代理地址为：${it.sockurl}")
                apputils.preparevpn(this@mainactivity,it.sockurl)
               // preparevpn()
            }
        }
    }
    flowbus.with<messageevent>("onstopvpn").register(this@mainactivity) {
        logutils.d(tag, "收到vpn消息为：$it")
        if (it.message == "stop" && !it.state) {
            apputils.stopvpn(this@mainactivity)
        }
    }
}