如何切换Kotlin StateFlow的数据源 实现MediatorLiveData同等动态过滤能力

原LiveData实现逻辑 #

原基于MediatorLiveData的代码作用是根据传入的过滤模式动态切换数据源，每次切换时会移除旧数据源的监听，避免旧数据回调：

class RosterMotor(private val repo: ToDoRepository) : ViewModel() {
  private val _states = MediatorLiveData<RosterViewState>()
  val states: LiveData<RosterViewState> = _states
  private var lastSource: LiveData<RosterViewState>? = null

  init {
    load(FilterMode.ALL)
  }

  fun load(filterMode: FilterMode) {
    lastSource?.let { _states.removeSource(it) }

    val items =
      repo.items(filterMode).map { RosterViewState(it, filterMode) }.asLiveData()

    _states.addSource(items) { viewstate ->
      _states.value = viewstate
    }

    lastSource = items
  }
...
}

现有实现的问题 #

你当前的实现存在旧协程泄露的问题：每次调用load方法都会启动一个新的协程收集数据源，之前的旧协程没有被取消，旧数据源仍然会继续发送数据覆盖新的过滤结果，和原MediatorLiveData的逻辑并不等价。

最优实现方案（推荐） #

使用flatMapLatest操作符实现自动切换数据源，该操作符会在新事件到来时自动取消之前的上游流收集，完全等效于原逻辑的removeSource+addSource逻辑，写法更简洁不易出错：

class RosterMotor(private val repo: ToDoRepository) : ViewModel() {
    // 存储当前过滤模式的状态流
    private val _currentFilter = MutableStateFlow(FilterMode.ALL)
    
    val states: StateFlow<RosterViewState> = _currentFilter
        // 过滤模式变化时自动切换数据源，旧流自动取消
        .flatMapLatest { filterMode ->
            repo.items(filterMode)
                .map { RosterViewState(it, filterMode) }
        }
        // 转换为持久化的StateFlow
        .stateIn(
            scope = viewModelScope,
            // 配置共享策略：无订阅者后5秒内再订阅不需要重启流，适配屏幕旋转等场景
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = RosterViewState()
        )

    fun load(filterMode: FilterMode) {
        // 仅需更新过滤模式即可自动触发数据源切换
        _currentFilter.value = filterMode
    }
}

备选兼容实现 #

如果要保留你原有手动启动协程的写法，需要增加对旧协程的取消逻辑，等效于原逻辑的移除旧数据源操作：

class RosterMotor(private val repo: ToDoRepository) : ViewModel() {
    private val _states = MutableStateFlow(RosterViewState())
    val states: StateFlow<RosterViewState> = _states
    // 存储上一次收集任务的Job
    private var lastLoadJob: Job? = null

    init {
        load(FilterMode.ALL)
    }

    fun load(filterMode: FilterMode) {
        // 取消上一次的收集任务，避免旧数据回调
        lastLoadJob?.cancel()
        lastLoadJob = viewModelScope.launch {
            repo.items(filterMode)
                .map { RosterViewState(it, filterMode) }
                .collect { _states.value = it }
        }
    }
}

内容的提问来源于stack exchange，提问作者Vivart