取消对信号的 kotlin 流收集答案

【问题标题】：Cancel kotlin flow collection on signal取消对信号的 kotlin 流收集
【发布时间】：2019-11-29 09:47:44
【问题描述】：

我正在努力为 Flow 创建一个“takeUntilSignal”运算符 - 一种在另一个流生成输出时将取消一个流的扩展方法。

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)
        }
    }
}

但这不起作用（并且使用了禁止使用的“withContext”方法，该方法被 Flow 明确删除以防止使用）。

编辑我将以下可憎的东西拼凑在一起，它不太符合定义（结果流只会在主流首次排放后取消），我觉得那里有更好的方法：

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()
    }

【问题讨论】：

为什么信号必须是流？如果它是 Deferred（单元素情况的自然选择），您将拥有 isCompleted 标志来检查 collect 的每次迭代。
嗨，Marko - 这是一个有趣的想法！我的具体用例是我有一个对话框 - 对话框中的字段绑定到流，我想在对话框关闭时取消这些流。当对话框关闭时，这些流不会发出，从而使这些集合处于一种不确定状态。我可以存储每个收集的作业并在收到关闭信号时取消所有作业，但这是很多簿记。
也许你误解了我的意思，我并没有暗示 async 的工作，而是一个简单的 CompletableDeferred 你明确地完成了一个值。关闭对话框将完成它，并且所有场流将检查它并在完成时自行关闭。
嗨，Marko - 如果我理解正确，它将类似于 collect { if (signal.isCompleted) throw CancellationException() else emit(it) }
是的，例如first() 的工作方式相同，通过抛出异常：try { collect { value -> result = value; throw AbortFlowException() } } catch (e: AbortFlowException) { // Do nothing }

标签： kotlin kotlin-coroutines

【解决方案1】：

使用couroutineScope并在里面启动新的协程：

fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit>): Flow<T> = flow {
    try {
        coroutineScope {
            launch {
                signal.take(1).collect()
                println("signalled")
                this@coroutineScope.cancel()
            }

            collect {
                emit(it)
            }
        }

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

【讨论】：

你确定吗？一个小测试确实有效。它是原始范围的子范围。你能指出我的描述，允许哪个范围。谢谢。
我在这里发现 kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/… 允许在内部 coroutineScope 中发射。
coroutineScope { } 等价于withContext(coroutineScope) { }，虽然后者被正式禁止，但这只是因为您可以使用任何上下文。所以这应该可行。
这不能编译：not enough information to infer type variable R
@IgorGanapolsky 此代码示例中没有变量 R。你能提供你的代码吗？

【解决方案2】：

查看https://github.com/hoc081098/FlowExt


package com.hoc081098.flowext

import com.hoc081098.flowext.internal.ClosedException
import com.hoc081098.flowext.internal.checkOwnership
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.launch

/**
 * Emits the values emitted by the source [Flow] until a [notifier] [Flow] emits a value or completes.
 *
 * @param notifier The [Flow] whose first emitted value or complete event
 * will cause the output [Flow] of [takeUntil] to stop emitting values from the source [Flow].
 */
public fun <T, R> Flow<T>.takeUntil(notifier: Flow<R>): Flow<T> = flow {
  try {
    coroutineScope {
      val job = launch(start = CoroutineStart.UNDISPATCHED) {
        notifier.take(1).collect()
        throw ClosedException(this@flow)
      }

      collect { emit(it) }
      job.cancel()
    }
  } catch (e: ClosedException) {
    e.checkOwnership(this@flow)
  }
}

【讨论】：