Kotlin 协程流中 RxJava .toList() 的等价物

Question

我有一种情况，我需要观察用户 ID，然后使用这些用户 ID 来观察用户。 userIds 或 users 可以随时更改，我希望让发出的用户保持最新。 这是我拥有的数据来源的示例：

data class User(val name: String)

fun observeBestUserIds(): Flow<List<String>> {
    return flow {
        emit(listOf("abc", "def"))
        delay(500)
        emit(listOf("123", "234"))
    }
}

fun observeUserForId(userId: String): Flow<User> {
    return flow {
        emit(User("${userId}_name"))
        delay(2000)
        emit(User("${userId}_name_updated"))
    }
}

在这种情况下，我希望排放量为：

[User(abc_name), User(def_name)]，然后

[User(123_name), User(234_name)]，然后

[User(123_name_updated), User(234_name_updated)]

我想我可以像这样在 RxJava 中实现这一点：

observeBestUserIds.concatMapSingle { ids ->
    Observable.fromIterable(ids)
        .concatMap { id ->
            observeUserForId(id)
        }
        .toList()
}

我会编写什么函数来生成一个发出它的流？

Answer 1

我相信您正在寻找combine，它为您提供了一个可以轻松调用toList() 的数组：

observeBestUserIds().collectLatest { ids ->
    combine(
        ids.map { id -> observeUserForId(id) }
    ) {
        it.toList()
    }.collect {
        println(it)
    } 
}

这是内部参数名称更明确的部分，因为您在 Stack Overflow 上看不到 IDE 的类型提示：

combine(
    ids.map { id -> observeUserForId(id) }
) { arrayOfUsers: Array<User> ->
    arrayOfUsers.toList()
}.collect { listOfUsers: List<User> ->
    println(listOfUsers)
}

输出：

[User(name=abc_name), User(name=def_name)]
[User(name=123_name), User(name=234_name)]
[User(name=123_name_updated), User(name=234_name)]
[User(name=123_name_updated), User(name=234_name_updated)]

Live demo（请注意，在演示中，所有输出都会同时出现，但这是演示站点的限制 - 代码在本地运行时，这些行的出现时间是您所期望的）

这避免了原始问题中讨论的 (abc_name_updated, def_name_updated)。但是，123_name_updated 和 234_name 仍然存在中间发射，因为首先发射 123_name_updated，它会立即发送组合版本，因为它们是每个流中的最新版本。

但是，这可以通过对排放进行去抖动来避免（在我的机器上，小至 1 毫秒的超时有效，但为了保守起见，我做了 20 毫秒）：

observeBestUserIds().collectLatest { ids ->
    combine(
        ids.map { id -> observeUserForId(id) }
    ) {
        it.toList()
    }.debounce(timeoutMillis = 20).collect {
        println(it)
    }
}

它会得到你想要的确切输出：

[User(name=abc_name), User(name=def_name)]
[User(name=123_name), User(name=234_name)]
[User(name=123_name_updated), User(name=234_name_updated)]

Live demo

Answer 2

不幸的是，在 kotlin Flow 的当前状态下，这不是微不足道的，似乎缺少重要的运算符。但请注意，您不是在寻找 rxJavas toList()。如果您尝试在 rxjava 中使用 toList 和 concatMap 来执行此操作，则必须等到所有 observabes 完成。 这不是你想要的。

不幸的是，我认为没有办法绕过自定义函数。

它必须汇总 observeUserForId 返回的所有结果，以获取您将传递给它的所有 id。它也不是一个简单的窗口函数，因为实际上可以想象一个observeUserForId 已经返回了两次，而另一个调用仍然没有完成。因此，检查您是否已经拥有与将 id 传递给聚合函数相同数量的用户是不够的，您还必须按用户 id 进行分组。

我会在今天晚些时候尝试添加代码。

编辑：正如这里所承诺的那样，我的解决方案是我冒昧地稍微增加了要求。因此，每次所有 userId 都具有值并且基础用户更改时，流程都会发出。我认为这更有可能是您想要的，因为用户可能不会同步更改属性。

如果这不是您想要的，请发表评论。

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


data class User(val name: String)

fun observeBestUserIds(): Flow<List<String>> {
    return flow {
        emit(listOf("abc", "def"))
        delay(500)
        emit(listOf("123", "234"))
    }
}

fun observeUserForId(userId: String): Flow<User> {
    return flow {
        emit(User("${userId}_name"))
        delay(2000)
        emit(User("${userId}_name_updated"))
    }
}

inline fun <reified K, V> buildMap(keys: Set<K>, crossinline valueFunc: (K) -> Flow<V>): Flow<Map<K, V>> = flow {
    val keysSize = keys.size
    val valuesMap = HashMap<K, V>(keys.size)
    flowOf(*keys.toTypedArray())
            .flatMapMerge { key -> valueFunc(key).map {v -> Pair(key, v)} }
            .collect { (key, value) ->
                valuesMap[key] = value
                if (valuesMap.keys.size == keysSize) {
                    emit(valuesMap.toMap())
                }
            }
}

fun observeUsersForIds(): Flow<List<User>> {
    return observeBestUserIds().flatMapLatest { ids -> buildMap(ids.toSet(), ::observeUserForId as (String) -> Flow<User>) }
            .map { m -> m.values.toList() }
}


fun main() = runBlocking {
    observeUsersForIds()
        .collect { user ->
            println(user)
        }
}

这将返回

[User(name=def_name), User(name=abc_name)]
[User(name=123_name), User(name=234_name)]
[User(name=123_name_updated), User(name=234_name)]
[User(name=123_name_updated), User(name=234_name_updated)]

可以在线运行代码here

Answer 3

您可以使用flatMapConcat

val users = observeBestUserIds()
        .flatMapConcat { ids ->
            flowOf(*ids.toTypedArray())
                .map { id ->
                    observeUserForId(id)
                }
        }
        .flattenConcat()
        .toList()

或

    observeBestUserIds()
        .flatMapConcat { ids ->
            flowOf(*ids.toTypedArray())
                .map { id ->
                    observeUserForId(id)
                }
        }
        .flattenConcat()
        .collect { user ->

        }