如何在Kotlin协程Flow中实现RxJava .toList()等效逻辑

Here's how you can achieve the desired behavior using Kotlin Flow operators:

observeBestUserIds()
    .flatMapLatest { userIds ->
        // Create a list of Flows, one for each userId
        val userFlows = userIds.map { userId -> observeUserForId(userId) }
        // Combine all user Flows to emit the latest list of Users
        combine(userFlows) { usersArray -> usersArray.toList() }
    }

Explanation of the Operators: #

1. flatMapLatest:

   • This operator replaces the current downstream Flow whenever the upstream (observeBestUserIds) emits a new list of user IDs.
   • When a new list arrives, it cancels the previous combined Flow (stopping observation of the old user IDs) and immediately starts observing the new set of users. This aligns with your need to switch to the latest user IDs without waiting for previous streams to complete.

2. combine:

   • The combine operator takes a collection of Flows and emits a new value whenever any of the input Flows emits a value.
   • For our use case, it waits for each user's Flow to emit at least once (getting the initial user data), then emits a list of the latest User objects in the same order as the input user IDs.
   • Whenever any user's data updates (their Flow emits a new value), combine will emit a fresh list containing all the latest user data for the current set of IDs.

How This Matches Your Expected Output: #

• First emission: When the initial user IDs (abc, def) are emitted, combine collects the first value from each user's Flow and outputs [User(abc_name), User(def_name)].
• Second emission: After 500ms, when the IDs switch to (123, 234), flatMapLatest cancels the old stream and combines the new user Flows, emitting [User(123_name), User(234_name)].
• Third emission: After another 1500ms (total 2000ms), each of the new user Flows emits their updated values. combine detects these changes and outputs [User(123_name_updated), User(234_name_updated)].

Comparison to Your RxJava Implementation: #

Your RxJava code uses concatMapSingle with toList(), which would only emit once per ID list (when all user Observables complete). In contrast, the Flow solution using flatMapLatest + combine continues to emit updates as user data changes, which matches your requirement for continuous output of the latest user information.

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