I wanted to know how can I send/emit items to a Kotlin.Flow, so my use case is:
在使用者/视图模型/演示者中，我可以subscribe使用collect函数:
fun observe() {
coroutineScope.launch {
// 1. Send event
reopsitory.observe().collect {
println(it)
}
}
}
But the issue is in the Repository side, with RxJava we could use a Behaviorsubject expose it as an Observable/Flowable and emit new items like this:
behaviourSubject.onNext(true)
但每当我建立一个新的流程:
flow {
}
我只能collect岁.如何将值发送到流？
如果您想获得订阅/Collection 的latest值，则应使用ConflatedBroadcastChannel:
private val channel = ConflatedBroadcastChannel<Boolean>()
这将复制BehaviourSubject，以将通道作为流公开:
// Repository
fun observe() {
return channel.asFlow()
}
Now to send an event/value to that exposed Flow simple send to this channel.
// Repository
fun someLogicalOp() {
channel.send(错误) // This gets sent to the ViewModel/Presenter and printed.
}
慰问:
错误
如果您只希望接收after个值，那么开始收集时应该使用BroadcastChannel.
Behaves as an Rx's PublishedSubject
private val channel = BroadcastChannel<Boolean>(1)
fun broadcastChannelTest() {
// 1. Send event
channel.send(符合事实的)
// 2. Start collecting
channel
.asFlow()
.collect {
println(it)
}
// 3. Send another event
channel.send(错误)
}
错误
Only 错误 gets printed as the first event was sent before collect { }.
Behaves as an Rx's BehaviourSubject
private val confChannel = ConflatedBroadcastChannel<Boolean>()
fun conflatedBroadcastChannelTest() {
// 1. Send event
confChannel.send(符合事实的)
// 2. Start collecting
confChannel
.asFlow()
.collect {
println(it)
}
// 3. Send another event
confChannel.send(错误)
}
符合事实的
错误
两个事件都会打印出来，您总是会得到latest的值(如果存在).
Also, want to mention Kotlin's team development on DataFlow (name pending):
哪个似乎更适合这个用例(因为它将是cold stream).