Kotlin Flow：当 Fragment 变得不可见时取消订阅 SharedFlow

Question

我读过类似的主题，但找不到正确的答案：

• How to end / close a MutableSharedFlow?
• Kotlin Flow: How to unsubscribe/stop
• StateFlow and SharedFlow. Making cold flows hot using shareIn - Android 文档
• Introduce SharedFlow - 由 Roman Elizarov 发起的 GH 讨论

在我的Repository 课程中，我有一个感冒Flow 我想分享给2 Presenters/ViewModels 所以我的选择是使用shareIn 运算符。

让我们看一下 Android 文档的示例：

val latestNews: Flow<List<ArticleHeadline>> = flow {
    ...
}.shareIn(
    externalScope,  // e.g. CoroutineScope(Dispatchers.IO)?
    replay = 1,
    started = SharingStarted.WhileSubscribed()
)

文档对externalScope 参数的建议：

用于共享流的 CoroutineScope。这个作用域的寿命应该比任何消费者都长，以使共享流在需要时保持活力。

但是，在寻找有关如何停止订阅 Flow 的答案时，第二个链接中投票最多的答案是：

解决方案不是取消流程，而是取消流程的启动范围。

对我来说，这些答案在SharedFlow 的案例中是矛盾的。不幸的是，我的Presenter/ViewModel 仍然收到最新数据，即使在其onCleared 被调用后。

如何预防？这是我如何在Presenter/ViewModel 中使用此Flow 的示例：

fun doSomethingUseful(): Flow<OtherModel> {
    return repository.latestNews.map(OtherModel)

如果这可能有帮助，我正在使用 MVI 架构，所以 doSomethingUseful 会对用户创建的一些意图作出反应。

Answer 1

我试图提供一个带有相关 cmets 的最小示例。如前所述，SharedFlow 的工作方式与 RxJava 中的 ConnectableObservable 非常相似。上游只会被订阅一次，这意味着计算只对冷的上游流进行一次。您的存储库什么都不做，因为它是一个冷流，在SharedFlow 订阅之前永远不会“收集”，因此它没有范围。

同时使用 RxJava 和 Flow 有很多相似之处。创建 Flow 和 Collector 接口似乎几乎没有必要，如果从基础 Reactive Streams 接口扩展而来，开发人员可以更轻松地进行转换 - 但我不知道根本原因 - 也许他们想要更多使用新的 api 实现灵活性，或者从 Java 9 实现和 RxJava 等另一个 Reactive Streams 实现中脱颖而出。

class MyViewModel : ViewModel(), CoroutineScope {

    override val coroutineContext: CoroutineContext = Dispatchers.Main.immediate + SupervisorJob() // optional + CoroutineExceptionHandler()

    private val latestNews: Flow<List<String>> = doSomethingUseful()
            .flowOn(Dispatchers.IO) // upstream will operate on this dispatch
            .shareIn(scope = this, // shared in this scope - becomes hot flow  (or use viewModelScope) for lifetime of your view model - will only connect to doSomethingUseful once for lifetime of scope
                     replay = 1,
                     started = SharingStarted.WhileSubscribed())


    fun connect() : Flow<List<String>> = latestNews // expose SharedFlow to "n" number of subscribers or same subscriber more than once

    override fun onCleared() {
        super.onCleared()
        cancel() // cancel the shared flow - this scope is finished
    }
}

class MainActivity : AppCompatActivity(), CoroutineScope {

    override val coroutineContext: CoroutineContext = Dispatchers.Main.immediate + SupervisorJob()

    private var job : Job? = null

    // supply the same view model instance on config changes for example - its scope is larger
    private val vm : MyViewModel by viewModels()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
    }

    override fun onStart() {
        super.onStart()

        job = launch {
            vm.connect().collect {
                // observe latest emission of hot flow and subsequent emissions if any - either reconnect or connect for first time
            }
        }
    }

    override fun onStop() {
        super.onStop()

        // cancel the job but latest news is still "alive" and receives emissions as it is running in a larger scope of this scope
        job?.cancel()
    }

    override fun onDestroy() {
        super.onDestroy()
        // completely cancel this scope - the ViewModel scope is unaffected
        cancel()
    }
}

Answer 2

感谢 Mark Keen 的 cmets 和帖子，我想我设法获得了令人满意的结果。

我了解shareIn 参数中定义的范围不必与我的消费者操作的范围相同。将BasePresenter/BaseViewModel 的范围从CoroutineScope 更改为viewModelScope 似乎解决了主要问题。您甚至不需要手动取消此范围，如 Android docs 中所定义：

init {
    viewModelScope.launch {
        // Coroutine that will be canceled when the ViewModel is cleared.
    }
}

请记住，默认的viewModelScope 调度程序是Main，这并不明显，可能不是您想要的！要更改调度程序，请使用viewModelScope.launch(YourDispatcher)。

更重要的是，我的热SharedFlow 是从另一个冷Flow 转换而来的，它是在callbackFlow 回调API 上创建的（它基于Channels API - 这很复杂......）

将集合范围更改为viewModelScope 后，我在从该API 发出新数据时收到ChildCancelledException: Child of the scoped flow was cancelled 异常。这个问题在 GitHub 上的两个问题中都有详细记录：

• kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
• SendChannel.offer should never throw

如上所述，使用offer 和send 的发射之间存在细微差别：

offer 用于非暂停上下文，而 send 用于暂停上下文。

不幸的是，就传播的异常而言，offer 是非对称的发送（通常会忽略来自 send 的 CancellationException，而来自在 nom-suspending 上下文中来自 offer 的 CancellationException 则不会）。

我们希望通过 offerOrClosed 或更改报价语义在 #974 中修复它

对于 1.4.2 的 Kotlin Coroutines，#974 还没有修复 - 我希望它会在不久的将来避免意外CancellationException。

最后，我建议在shareIn 运算符中使用started 参数。在所有这些更改之后，我不得不在我的用例中从 WhileSubscribed() 更改为 Lazily。

如果我发现任何新信息，我会更新这篇文章。希望我的研究能节省一些人的时间。

Answer 3

使用共享流。在下面的示例中，我从一个片段发出值并在另一个片段上收集它。

视图模型：

class MenuOptionsViewModel : ViewModel() {
private val _option = MutableSharedFlow<String>()
val option = _option.asSharedFlow()

suspend fun setOption(o : String){
    _option.emit(o)
}
}

片段发射值：

class BottomSheetOptionsFragment  : BottomSheetDialogFragment() , KodeinAware{

    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    super.onViewCreated(view, savedInstanceState)
    menuViewModel = activity?.run {
        ViewModelProviders.of(this).get(MenuOptionsViewModel::class.java)
    } ?: throw Exception("Invalid Activity")

    listViewOptions.adapter = ArrayAdapter<String>(
        requireContext(),
        R.layout.menu_text_item,
        options
    )

    listViewOptions.setOnItemClickListener { adapterView, view, i, l ->
        val entry: String = listViewOptions.getAdapter().getItem(i) as String

// here we are emitting values
        GlobalScope.launch { menuViewModel.setOption(entry) }
        Log.d(TAG, "emitting flow $entry")
        dismiss()
    }
}
}

片段收集值：

class DetailFragment : BaseFragment(), View.OnClickListener, KodeinAware,
OnItemClickListener {

override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        super.onViewCreated(view, savedInstanceState)
        menuViewModel = activity?.run {
            ViewModelProviders.of(this).get(MenuOptionsViewModel::class.java)
        } ?: throw Exception("Invalid Activity")


// collecting values
                lifecycleScope.launchWhenStarted {
            menuViewModel.option.collect {
                Log.d(TAG, "collecting flow $it")
                
            }
        }
}