在不取消当前范围的情况下退订 Kotlin Flow

Question

我在 Android 的 repeatOnLifecycle 块内订阅 Kotlin SharedFlow 布尔值。我想订阅，直到收到第一个 true 布尔值并采取行动。

一旦收到第一个 true 布尔值，我需要取消订阅并在生命周期范围内运行处理函数，以便在生命周期转换为无效状态时取消它。

当我在当前范围上调用取消并将处理代码嵌入NonCancellable 上下文时，它不会在生命周期事件中被取消。

我想我想要takeWhile 之类的东西，包括与谓词不匹配的第一个元素。

下面是一些示例流程，我想在其中收集所有元素，直到 $ 符号：

true $ ...
false, true $ ...
false, false, true $ ...

示例代码：

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        flowOfInterest.collectLatest {
            if (it) {
                stopCollection()
                doOnTrue()
            } else {
                doOnFalse()
            }
        }
    }
}

实现此行为的正确/最简单方法是什么？

谢谢！

Answer 1

您可以使用first 函数继续收集，直到给定谓词返回true。

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        flowOfInterest.first { 
            if(it) doOnTrue() else doOnFalse()
            it 
        }
    }
}

Answer 2

在这里回答我自己的问题。我需要的是一个类似于takeWhile 的运算符，其中包含第一个不匹配的元素。可以使用transformWhile 运算符创建这样的运算符，如下所示：

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        flowOfInterest.transformWhile {
            emit(it)
            !it
        }.collectLatest {
            if (it) {
                doOnTrue()
            } else {
                doOnFalse()
            }
        }
    }
}

这不像我希望的那么好和紧凑，但它确实有效。

编辑：或者，您可以使用 Arpit Shukla 的答案并在“第一个”函数的谓词中执行操作。

Answer 3

你可以在repeatOnLifecycle 函数范围内启动一个新的协程，因为launch 函数返回一个Job，所以你可以在以后使用它来取消你的工作。

此外，如果你想在repeatOnLifecycle 范围内监听多个流，你需要启动一个子协程让它们并行运行。所以它们不会互相阻塞，取消一个流不会影响其他流。

示例代码：

lateinit var job: Job

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        job = launch {
            flowOfInterest.collectLatest {
                if (it) {
                    stopCollection()
                    doOnTrue()
                } else {
                    doOnFalse()
                }
            }
        }
    }
}

func stopCollection() {
    job.cancel()
}

Answer 4

您还可以使用takeWhile 定义控制从流中收集的条件。

 flowOfInterest.takeWhile { !it }.collect {
     //execute smth
 }

如果你的流在达到状态后是“冷的”，当条件不满足时，如果没有更多的观察者，流将完全停止发射。否则，它将继续（例如，如果您使用共享流），但您当前的观察者（您应用于问题）将停止收集。

作为针对您的特定情况的解决方法，如果您的流程不是“冷”，您可以致电

flowOfInterest.take(1).collect { ... } 

接收谓词返回 false 的元素。

您还可以使用额外的布尔值，当您达到某些条件时，您将在收集块中修改该值，这将是takeWhile 的谓词。