PublishSubject with Kotlin coroutines (Flow)

Flow is a cold asynchronous stream, just like an Observable.

All transformations on the flow, such as map and filter do not trigger flow collection or execution, only terminal operators (e.g. single) do trigger it.

The onEach method is just a transformation. Therefore you should replace it with the terminal flow operator collect. Also you could use a BroadcastChannel to have cleaner code:

private val channel = BroadcastChannel<Boolean>(1)suspend fun someMethod(b: Boolean) {    channel.send(b)}suspend fun observe() {  channel    .asFlow()    .debounce(500)    .collect {        // value received    }}

_{Update: At the time the question was asked there was an overload of debounce with two parameters (like in the question). There is not anymore. But now there is one which takes one argument in milliseconds (Long).}

It should be SharedFlow/MutableSharedFlow for PublishProcessor/PublishRelay

private val _myFlow = MutableSharedFlow<Boolean>(                          replay = 0,                          extraBufferCapacity = 1, // you can increase                                BufferOverflow.DROP_OLDEST)val myFlow = _myFlow.asSharedFlow()// ...fun someMethod(b: Boolean) {    _myFlow.tryEmit(b)}fun observe() {    myFlow.debounce(500)          .onEach {  }          // flowOn(), catch{}          .launchIn(coroutineScope)}

And StateFlow/MutableStateFlow for BehaviorProcessor/BehaviorRelay.

private val _myFlow = MutableStateFlow<Boolean>(false)val myFlow = _myFlow.asStateFlow()// ...fun someMethod(b: Boolean) {    _myFlow.value = b // same as _myFlow.emit(v), myFlow.tryEmit(b)}fun observe() {    myFlow.debounce(500)          .onEach {  }          // flowOn(), catch{}          .launchIn(coroutineScope)}

StateFlow must have initial value, if you don't want that, this is workaround:

private val _myFlow = MutableStateFlow<Boolean?>(null)val myFlow = _myFlow.asStateFlow()                    .filterNotNull()

MutableStateFlow uses .equals comparison when setting new value, so it does not emit same value again and again (versus distinctUntilChanged which uses referential comparison).

So MutableStateFlow ≈ BehaviorProcessor.distinctUntilChanged(). If you want exact BehaviorProcessor behavior then you can use this:

private val _myFlow = MutableSharedFlow<Boolean>(                              replay = 1,                               extraBufferCapacity = 0,                              BufferOverflow.DROP_OLDEST)

ArrayBroadcastChannel in Kotlin coroutines is the one most similar to PublishSubject.

1. Like PublishSubject, an ArrayBroadcastChannel can have multiplesubscribers and all the active subscribers are immediately notified.
2. Like PublishSubject, events pushed to this channel are lost, if there are no active subscribers at the moment.

Unlike PublishSubject, backpressure is inbuilt into the coroutine channels, and that is where the buffer capacity comes in. This number really depends on which use case the channel is being used for. For most of the normal use cases, I just go with 10, which should be more than enough. If you push events faster to this channel than receivers consuming it, you can give more capacity.