取消信号上的kotlin流收集

Question

我正在为一个流创建一个'take untilsignal'运算符--一个扩展方法，当另一个流生成输出时，它会取消一个流。

fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit>): Flow<T>

我最初的努力是尝试在与主要流收集相同的协同范围内启动信号流的收集，并取消协同范围：

fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit>): Flow<T> = flow {
    kotlinx.coroutines.withContext(coroutineContext) {
        launch {
            signal.take(1).collect()
            println("signalled")
            cancel()
        }
        collect {
            emit(it)
        }
    }
}

编辑：我拼凑了以下这些讨厌的东西，它不太符合定义（结果流只会在第一次从主流发射后取消），我觉得有一个更好的方法：

fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit>): Flow<T> =
    combine(
        signal.map { it as Any? }.onStart { emit(null) }
    ) { x, y -> x to y }
        .takeWhile { it.second == null }
        .map { it.first }

edit2再试一次，使用ChannelFlow：

fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit>): Flow<T> =
    channelFlow {
        launch {
            signal.take(1).collect()
            println("hello!")
            close()
        }
        collect { send(it) }
        close()
    }

Answer 1

使用couroutineScope并在内部启动新的coroutine：

fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit>): Flow<T> = flow {
    try {
        coroutineScope {
            launch {
                signal.take(1).collect()
                println("signalled")
                [e-mail]()
            }

            collect {
                emit(it)
            }
        }

    } catch (e: CancellationException) {
        //ignore
    }
}