android - 运行一次 kotlin 流，但在下游接收两次

Question

场景

我有一个热流 EventHandler.sharedFlow 在单击按钮时发出。

流由在 OnEach{} 中执行某些操作的 Repository 接收。

存储库流然后由两个事件收集器 EventCollectorA 和 EventCollectorB 接收。

然后将事件收集器流合并并收集到 MyViewModel 中。

问题

两个事件收集器导致 onEach{...} 在每次点击时运行两次。但是我只想运行一次 onEach{...} 并在两个事件收集器中接收它。我怎样才能做到这一点？

注意:我使用 Hilt 只有一个实例 Repository、EventCollectorA 和 EventCollectorB

代码

@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        val binding = ActivityMainBinding.inflate(layoutInflater)
        setContentView(binding.root)

        val viewModel = ViewModelProvider(this).get(MyViewModel::class.java)
        binding.buttonB.setOnClickListener {
            viewModel.userClickEvent("Click Event")
        }
    }
}

@HiltViewModel
class MyViewModel @Inject constructor(
    private val eventHandler: EventHandler,
    private val eventCollectorA: EventCollectorA,
    private val eventCollectorB: EventCollectorB,
) : ViewModel() {
    fun userClickEvent(event: String) = viewModelScope.launch {
        eventHandler.userClick(event)
    }

    init {
        viewModelScope.launch {
            combine(
                eventCollectorA.sharedFlow,
                eventCollectorB.sharedFlow
            ) { a, b ->
                {/*do something*/}
            }.collect()
        }
    }
}

class EventHandler  {
    private val _sharedFlow = MutableSharedFlow<String>()
    val sharedFlow = _sharedFlow.asSharedFlow()

    suspend fun userClick(event: String) {
        _sharedFlow.emit(event)
    }
}

class Repository constructor(
    eventHandler: EventHandler,
) {
    val sharedFlow = eventHandler.sharedFlow
            .filter { it == "Click Event" }
            .onEach {/*do something*/} /*onEach is called twice on click event. I only want it called once*/ 
            .onStart { emit("Begin") }
}

class EventCollectorA constructor(repository: Repository) {
    val sharedFlow = repository.sharedFlow.map {
        it
    }
}

class EventCollectorB constructor(repository: Repository) {
    val sharedFlow = repository.sharedFlow.map {
        it
    }
}

Answer 1

这里的问题是，虽然 eventHandler.sharedFlow 是一个 SharedFlow，但在对其应用任何运算符之后，我们会得到一个常规的非共享流。 filter()、onEach() 和 onStart() 分别针对每个新集合运行。如果你想在集合之间共享它们，你需要在应用它们之后构造另一个共享流:

    val sharedFlow = eventHandler.sharedFlow
            .filter { it == "Click Event" }
            .onEach {/*do something*/}
            .onStart { emit("Begin") }
            .shareIn(...)   

进一步说明

我们需要注意，常规的冷流与实时数据流不同。它更像是此类流的来源，对于每个新集合，我们都会启动全新的数据流。例如，如果我们使用 flow { } 构建器创建一个流，我们只有一个流对象，但是如果我们对其多次调用 collect {}，那么对于每个集合都会一次又一次地调用 lambda。同样，我们用于构建新流的每个运算符也会针对每个集合单独调用。

您可以将 shareIn() 视为创建一个服务，该服务观察其上游流并将其数据复制到其每个下游流。无论我们收集多少次共享流量，上游流量只会收集一次。 shareIn() 之上的操作符将被调用一次，而 shareIn() 下面的操作符将针对每个集合单独调用。