只有Kotlin协同程序的数据流？

Posted 2021-05-05

使用RxJava，我已经习惯了我的存储库返回Observables数据，这些数据会在发生潜在变化时自动更新。我通过简单地在我的存储库中有一个主题来获得相关的变更信息，并且像getAll()这样的可观察者就可以了。

举个例子，把这个psuedo代码看作代码片段：

fun getAll(): Observable<List<Model> {
    subject
        .filter { isChangeRelevant(it) }
        .startWith(initialChangeEvent)
        .map { queryAll() }
}

---

我一直很好奇如何使用协程只能实现同样的事情？

答案

你可以使用Kotlin Coroutines Channels。

如果你只想让你的值像流一样发出（所以你可以用它来for-each）你可以使用produce来创建它们（它返回一个ReceiveChannel）：

fun test(): ReceiveChannel<Int>{
        return produce {
            send(1)
            send(5)
            send(100)
        }
    }

您可以在consumeEach的值上使用for-each（或test()）来接收其值。

如果您希望您的频道与RxJava的PublishSubject完全相同，则可以使用ConflatedBroadCastChannel，并向其发出值：

val broadCastChannel = ConflatedBroadcastChannel<Int>()

您可以使用broadCastChannel.offer(value)将值发送到频道。

要从通道接收值，您可以使用简单的for-each循环：

for (i in broadCastChannel.openSubscription()) {
       //your values
}