PublishSubject with Kotlin coroutines (Flow)
Flow
is a cold asynchronous stream, just like an Observable
.
All transformations on the flow, such as
map
andfilter
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
)