Is there analogue for RxJava Subject in Kotlin Coroutines?

Question

In 2020 a lot of android developers are talking about Kotlin Coroutines. I'm trying to understand it and how coroutines can help me in my project.

So my question: is there analogue in Coroutines for RxJava Subjects? (As minimum for PublishSubject).

What I want - I use PublishSubject for sending events from ViewModel to my View. I subscribe to eventsSubject on onStart() method and dispose on onStop() method.

So the minimal requirements for Kotlin Coroutines analogue are:

• Easy testing (I use TestSubscriber and it is awesome)
• I want to send events without buffering
• Easy to subscribe/unsubscribe

There is sample of my use case:

ViewModel:

abstract class AbsStateViewModel<State, Event> : AbsViewModel() {
    private val stateSubject = BehaviorSubject.create<State>()
    private val eventSubject = PublishSubject.create<Event>()

    protected val requireState: State
        get() = stateSubject.value!!

    fun getStateObservable(): Observable<State> = stateSubject

    fun getEventObservable(): Observable<Event> = eventSubject

    protected fun sendEvent(event: Event) {
        eventSubject.onNext(event)
    }

    protected fun setState(state: State) {
        stateSubject.onNext(state)
    }
}

And usages:

viewModel.getEventObservable() // called on onAttach()
            .subscribe(
                    this::handleEvent,
                    this::defaultHandleException
            )
            .disposeOnDetach() // my extensions 

Answer 1

yes in coroutines there are the analogue of rx subjects, the channels. If you want to reproduce the behavior of PublishSubject you can use the BroadcastChannel else if you want to reproduce the behavior of BehaviorSubject you can use the ConflatedBroadcastChannel.