如何等待流程完成发出值

Question

我的存储库中有一个函数“getUser”，它根据提供的 id 发出一个代表用户的对象。

flow function

fun getUser(id: String) = callbackFlow {
    val collectionReference: CollectionReference =
        FirebaseFirestore.getInstance().collection(COLLECTION_USERS)

    val query: Query = collectionReference.whereEqualTo(ID, id)

    query.get().addOnSuccessListener {
        val lst = it.toObjects(User::class.java)
        if (lst.isEmpty())
            offer(null)
        else
            offer(it.toObjects(User::class.java)[0])
    }

    awaitClose()
}

我在另一个类中需要这些值。我遍历一个 id 列表，并将收集到的用户添加到一个新列表中。我如何在收集值时等待列表完成，然后再调用 return？

collector function

private fun computeAttendeesList(reminder: Reminder): ArrayList<User> {
    val attendeesList = arrayListOf<User>()
    for (friend in reminder.usersToShare) {
        repoScope.launch {
            Repository.getUser(friend).collect {
                it?.let { user ->
                    if (!attendeesList.contains(user))
                        attendeesList.add(user)
                }
            }
        }
    }
    return attendeesList
}

我不想使用实时数据，因为这不是与 UI 相关的类。

Answer 1

这段代码中有多个问题需要解决：

1. getUser() 旨在返回一个 User，但它目前返回一个 Flow<User> 这永远不会结束，并且永远不会返回一个以上的用户。
2. 从多个并发查询构造用户列表的方式不是线程安全的（因为在多线程 IO 调度器上执行多个launches，它们都直接更新同一个不安全列表）
3. 实际用例是从 Firebase 获取用户列表，但使用单个 ID 的多个查询而不是单个查询

#1 的解决方案

让我们先解决#1。这是getUser() 的一个版本，它暂停单个User 而不是返回Flow：

suspend fun getUser(id: String): User {
    val collectionReference = FirebaseFirestore.getInstance().collection(COLLECTION_USERS)
    val query = collectionReference.whereEqualTo(ID, id)

    return query.get().await().let { it.toObjects(User::class.java) }.firstOrNull()
}

// use the kotlinx-coroutines-play-services library instead
private suspend fun <T> Task<T>.await(): T {
    return suspendCancellableCoroutine { cont ->
        addOnCompleteListener {
            val e = exception
            if (e == null) {
                @Suppress("UNCHECKED_CAST")
                if (isCanceled) cont.cancel() else cont.resume(result as T)
            } else {
                cont.resumeWithException(e)
            }
        }
    }
}

事实证明，这个await() 函数已经被编写（以更好的方式）并且在kotlinx-coroutines-play-services library 中可用，因此您实际上不需要自己编写。

#2 的解决方案

如果我们不能按照#3重写整个事情，我们可以这样处理问题#2：

private suspend fun computeAttendeesList(reminder: Reminder): List<User> {
    return reminder.usersToShare
        .map { friendId -> 
            repoScope.async { Repository.getUser(friendId) }
        }
        .map { it.await() }
        .toList()
}

#3 的解决方案

相反，我们可以直接向 Firebase 查询整个列表：

suspend fun getUsers(ids: List<String>): List<User> {
    val collectionReference = FirebaseFirestore.getInstance().collection(COLLECTION_USERS)
    val query = collectionReference.whereIn(ID, ids)

    return query.get().await().let { it.toObjects(User::class.java) }
}

然后以非常基本的方式消费它：

private suspend fun computeAttendeesList(reminder: Reminder): List<User> {
    return Repository.getUsers(reminder.usersToShare)
}

或者，您可以使此函数阻塞（删除suspend）并将您的调用包装在runBlocking 中（如果您确实需要阻塞当前线程）。

请注意，此解决方案未强制执行任何调度程序，因此如果您需要特定范围或调度程序，可以使用 withContext 包装其中一个挂起函数调用。