Kotlin实现RxBus3

Posted 2022-11-10 Steven Jon

1、添加依赖

    
    implementation "io.reactivex.rxjava3:rxjava:3.1.2"
    implementation "io.reactivex.rxjava3:rxandroid:3.0.0"

2、RxBus3.kt

class RxBus3 private constructor()

    private val subjectMapper = ConcurrentHashMap<Any, MutableList<Subject<Any>>>()

    private val subscriber = ConcurrentHashMap<Any, ConcurrentHashMap<Any, Observable<*>>>()

    private val subscriptionMapper = ConcurrentHashMap<Any, MutableList<Disposable>>()

    companion object 
        val instance : RxBus3 by lazy(mode = LazyThreadSafetyMode.SYNCHRONIZED) 
            RxBus3()
        
    

    private fun isEmpty(collection: Collection<Subject<*>?>?): Boolean 
        return null == collection || collection.isEmpty()
    

    fun register(@NonNull tag: Any, @NonNull consumer: Consumer<Any>, clazz: Class<*>): Observable<Any> 
        var subjectList = subjectMapper[tag]
        if (null == subjectList) 
            subjectList = ArrayList()
            subjectMapper[tag] = subjectList
        
        val subject: Subject<Any>
        subject = PublishSubject.create()
        subjectList.add(subject)

        val sub = subject.observeOn(AndroidSchedulers.mainThread())
            .subscribe(consumer)  obj: Throwable -> obj.printStackTrace() 
        var subList: MutableList<Disposable>? = subscriptionMapper[clazz]
        if (null == subList) 
            subList = ArrayList()
        
        subList.add(sub)
        subscriptionMapper[clazz] = subList

        var subMap: ConcurrentHashMap<Any, Observable<*>>? = subscriber[clazz]
        if (subMap == null) 
            subMap = ConcurrentHashMap()
        
        subMap[tag] = subject
        subscriber[clazz] = subMap

        return subject
    


    fun unregister(clazz: Class<*>) 
        unsubscribe(clazz)
        val sub: ConcurrentHashMap<Any, Observable<*>>? = subscriber[clazz]
        if (sub != null) 
            for (tag in sub.keys) 
                val value = sub[tag]
                if (null != value) 
                    unregister(tag)
                
            
        

    

    private fun unsubscribe(clazz: Class<*>) 
        val sub = subscriptionMapper.remove(clazz)
        if (sub != null && sub.isNotEmpty()) 
            val it = sub.listIterator()
            while (it.hasNext()) 
                val subscriptDispose = it.next()
                subscriptDispose.dispose()
                it.remove()
            
        
    

    private fun unregister(@NonNull tag: Any) 
        subjectMapper[tag]?.remove(tag)
    

    fun post(@NonNull tag: Any, @NonNull content: Any) 
        val subjectList = subjectMapper[tag]
        if (!isEmpty(subjectList)) 
            if (subjectList != null) 
                for (subject in subjectList) 
                    subject.onNext(content)
                
            
        
    

3、使用

// 1、根据tag发送事件
RxBus3.instance.post("tag_any_string", "content_xxx")
RxBus3.instance.post("tag_any_object", obj)
....

//2、注册tag事件源,并指定class
RxBus3.instance.register("tag_any_object", 
                obj -> run  println(obj) 
        , javaClass)

//3、删除指定class里面的所有事件
RxBus3.instance.unregister(javaClass)

非常感谢你能看到最后，如果能够帮助到你，是我的荣幸！