如何用多个流包装单个回调答案

【问题标题】：How to wrap single callback with multiple flows如何用多个流包装单个回调
【发布时间】：2019-10-25 08:49:27
【问题描述】：

我有一个第三方库，您可以在其中订阅一些 topic 以接收关于该 topic 的更新（带有回调）。但问题是，当您再次使用相同的 topic 调用 subscribe 方法时，之前的订阅会默默地替换为新的订阅。

现在，我想将回调替换为 Flow。于是我写了如下代码：

val flow: Flow<Message> = callbackFlow {
    lib.subscribe(topic) { message -> offer(message) }
    awaitClose { lib.unsubscribe(topic) }
}

但是当我多次拨打collect 时，只有最后一个收到消息。

那么我怎样才能做到这一点：

对同一主题有多个订阅
当最后一个订阅者关闭/取消时，取消订阅给定主题。

我已经看过BroadcastChannel，但是我找不到任何东西来解决第二个要求。而且我刚刚开始探索Flows，所以也许有一些我只是错过的简单包装器。

【问题讨论】：

标签： kotlin-coroutines

【解决方案1】：

如果您将Flow 转换为SharedFlow（当然，您仍然可以将其公开为更抽象的Flow 类型），那么后续订阅者将被添加而无需重新执行传递给@ 的块987654325@，表示该库的原订阅者不会被替换。

Kotlin Flow 库提供了一个Flow<T>.shareIn(): SharedFlow<T> 扩展，这使得转换为SharedFlow 变得非常容易。

这个答案还有另一个重要部分。当您调用shareIn() 时，您需要将SharingStarted.WhileSubscribed 作为参数之一传递。这在您的用例中至关重要，因为没有它，awaitClose() 可能不会被调用，这意味着您不会从库中退订。对于WhileSubscribed，完全取消订阅SharedFlow 也会导致取消订阅原始Flow，这将触发对awaitClose() 的调用。

您的代码可能如下所示：

val flow: Flow<Message> = callbackFlow {
    lib.subscribe(topic) { message -> offer(message) }
    awaitClose { lib.unsubscribe(topic) }
}.shareIn(coroutineScope, SharingStarted.WhileSubscribed(), replay = 0)

来自shareIn()上的文档：

WhileSubscribed() — 当第一个订阅者出现时启动上游流，当最后一个订阅者消失时立即停止...

【讨论】：