RxJava实现的EventBus

BaseActivity.kt

open class BaseActivity: AppCompatActivity() {

    private val subscriptions = CompositeSubscription()
    
    override fun onStart() {
        super.onStart()
        subscriptions.add(RxBus.toObserverable()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { onRxBusEvent(it)})
        subscriptions.add(RxBus.toStickyObserverable()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { onRxBusStickyEvent(it) })
    }

    override fun onStop() {
        super.onStop()
        subscriptions.clear()
    }
    
    override fun onDestroy() {
        super.onDestroy()
        subscriptions.unsubscribe()
    }

    protected open fun onRxBusEvent(event: Any) {
        
    }

    protected open fun onRxBusStickyEvent(event: Any) {

    }
}

RxBus.kt

object RxBus {

    private val bus = SerializedSubject<Any, Any>(PublishSubject.create())
    private val stickyBus = SerializedSubject<Any, Any>(ReplaySubject.create(1))

    fun send(event: Any) = bus.onNext(event)

    fun sendSticky(event: Any) = stickyBus.onNext(event)

    fun toObserverable(): Observable<Any> = bus

    fun toStickyObserverable(): Observable<Any> = stickyBus
}