如何在流中使用回调流？

Question

我正在尝试将 callbackFlow 包装在外部 flow 中 - 我想从外部流中发出一些项目，但我有一个旧的回调接口，我想适应 Kotlin 流程。我查看了几个examples of usage of callbackFlow，但我不知道如何在另一个流程中正确触发它。

这是一个例子：

class Processor {
    fun start(processProgress: ProcessProgressListener) {
        processProgress.onFinished() //finishes as soon as it starts!
    }
}

interface ProcessProgressListener {
    fun onFinished()
}

//main method here:
 fun startProcess(processor: Processor): Flow<String> {
     val mainFlow = flow {
         emit("STARTED")
         emit("IN_PROGRESS")
     }

     return merge(processProgressFlow(processor), mainFlow)
 }

 fun processProgressFlow(processor: Processor) = callbackFlow {
     val listener = object : ProcessProgressListener {
         override fun onFinished() {
             trySend("FINISHED")
         }
     }

     processor.start(listener)
 }

Processor 带有一个侦听器，该侦听器在进程完成时触发。发生这种情况时，我想发出最后一个项目FINISHED。

我调用整个流程的方式如下：

     runBlocking {
         startProcess(Processor()).collect {
             print(it)
         }
     }

但是，我没有得到任何输出。但是，如果我不使用megre 并且只返回mainFlow，我会得到STARTED 和IN_PROGRESS 项目。

我做错了什么？

Answer 1

您忘记在callbackFlow 块末尾调用awaitClose：

fun processProgressFlow(processor: Processor) = callbackFlow<String> {
    val listener = object : ProcessProgressListener {
        override fun onFinished() {
            trySend("FINISHED")
            channel.close()
        }
    }

    processor.start(listener)

    /*
     * Suspends until 'channel.close() or cancel()' is invoked
     * or flow collector is cancelled (e.g. by 'take(1)' or because a collector's coroutine was cancelled).
     * In both cases, callback will be properly unregistered.
     */
    awaitClose { /* unregister listener here */ }
}

awaitClose {} 应该用在callbackFlow 块的末尾。 否则，在外部取消的情况下，回调/侦听器可能会泄漏。

根据callbackFlow docs：

awaitClose 应该用于保持流程运行，否则在阻塞完成时通道将立即关闭。 awaitClose 参数在流消费者取消流收集或基于回调的 API 手动调用 SendChannel.close 时调用，通常用于在完成后清理资源，例如取消注册回调。使用awaitClose 强制是为了防止流收集被取消时内存泄漏，否则即使流收集器已经完成，回调也可能继续运行。为避免此类泄漏，如果块返回，但通道尚未关闭，此方法将抛出 IllegalStateException。