Kotlin流: How to unsubscribe/stop

Question

更新协程1.3.0-RC

工作版本:

@FlowPreview
suspend fun streamTest(): Flow<String> = channelFlow {
    listener.onSomeResult { result ->
        if (!isClosedForSend) {
            offer(result)
        }
    }

    awaitClose {
        listener.unsubscribe()
    }
}

还可以查看Roman Elizarov撰写的这篇中型文章: Callbacks and Kotlin Flows

原始问题

我有一个发出多个字符串的流:

@FlowPreview
suspend fun streamTest(): Flow<String> = flowViaChannel { channel ->
    listener.onSomeResult { result ->
            if (!channel.isClosedForSend) {
                channel.sendBlocking(result)
            }
    }
}

一段时间后，我想取消订阅该信息流。目前，我正在执行以下操作:

viewModelScope.launch {
    beaconService.streamTest().collect {
        Timber.i("stream value $it")
        if(it == "someString")
            // Here the coroutine gets canceled, but streamTest is still executed
            this.cancel() 
    }
}

如果协程被取消，该流仍将执行。没有订阅者在听新值。如何取消订阅并停止 stream函数？

Answer 1

使用最新版本的协程/Flows(1.2.x)，我现在不是一个好的解决方案。使用onCompletion，您将在流程停止时得到通知，但是此时您不在streamTest函数的范围内，因此很难停止监听新事件。

beaconService.streamTest().onCompletion {

}.collect {
    ...
}

在下一版的协程(1.3.x)中，这将非常容易。不推荐使用 flowViaChannel函数，而推荐使用 channelFlow 。此功能使您可以等待流关闭并在此时进行一些操作，例如。删除监听器:

channelFlow<String> {
    println("Subscribe to listener")

    awaitClose {
        println("Unsubscribe from listener")
    }
}