从 Firebase 监听器发出状态 - Kotlin Flow答案

【问题标题】：Emit State from Firebase listener - Kotlin Flow从 Firebase 监听器发出状态 - Kotlin Flow
【发布时间】：2020-09-21 10:56:13
【问题描述】：

我想使用 Kotlin Flow 来处理 FirebaseAuth 状态。我知道下面的代码是错误的，但我不知道如何修复它。我试过channelFlow，当我想要send或offer时它总是崩溃

   fun registerFlow(email: String, password: String) = flow {
    emit(AuthState.Loading)
    firebaseAuth.createUserWithEmailAndPassword(email, password)
        .addOnCompleteListener { task ->
            if (task.isSuccessful) {
                CoroutineScope(Dispatchers.IO).launch {      
                    emit(AuthState.Success(task.result?.user))
            } }else {
                  emit(AuthState.Error(task.exception))
            }
        }
}

}

Listener 内的协程抛出

z E/AndroidRuntime: FATAL EXCEPTION: DefaultDispatcher-worker-1
Process: pl.rybson.musicquiz, PID: 26578
java.lang.IllegalStateException: Flow invariant is violated:
        Emission from another coroutine is detected.
        Child of StandaloneCoroutine{Active}@4903a45, expected child of StandaloneCoroutine{Completed}@988059a.
        FlowCollector is not thread-safe and concurrent emissions are prohibited.
        To mitigate this restriction please use 'channelFlow' builder instead of 'flow'

我使用send()时的错误

FATAL EXCEPTION: DefaultDispatcher-worker-1
Process: pl.rybson.musicquiz, PID: 27105
kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed

【问题讨论】：

标签： android kotlin kotlin-coroutines

【解决方案1】：

flow 在代码块运行完成时结束。

您可以使用integration 使其成为挂起函数，而不是使用回调。

fun registerFlow(email: String, password: String) = flow {
    emit(AuthState.Loading)
    val result = firebaseAuth.createUserWithEmailAndPassword(email, password).await()
    if (result.isSuccessful) {     
        emit(AuthState.Success(result.result?.user))
    } else {
        emit(AuthState.Error(result.exception))
    }
}

【讨论】：

但请记住，这是一次性操作，不会持续更新。但在这个用例中，这不是问题

【解决方案2】：

方法一（推荐）

将此依赖项添加到您应用的 build.gradle 文件中：

implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-play-services:1.4.1'

此依赖项使您可以访问两个扩展功能：

Deferred<T>.asTask(): Task<T>
Task<T>.await(): T

这种方法有助于消除样板代码：

suspend fun getUser(id: String): User = 
        firebaseStorage
            .getReference("users/$id")
            .await()

方法二

如错误中所述，您应该改用 channelFlow。在撰写本文时，它仍处于试验阶段，但运行良好。

使用 firebase 的 addSuccessListener 和 channelFlow 的示例：

    @ExperimentalCoroutinesApi
    fun getStorageUrlFromRemoteFile(remoteFile: RemoteFile): Flow<Uri> = channelFlow {
        firebaseStorage.getReference(remoteFile.storagePath).downloadUrl
            .addOnSuccessListener {
                launch {
                    Timber.d("Sending uri in downloadRemoteFile() as $it")
                    send(it)
                    close()
                }
            }

        awaitClose()
    }

【讨论】：

【解决方案3】：

You can not change the context within 流程构建器。你可以做的是当你调用你的方法 registerFlow() 是使用 flowOn() 并提供正确的执行上下文。在你的情况下，你会有这样的事情：

registerFlow(email, password).flowOn(Dispatchers.Default).collect {}

【讨论】：

【解决方案4】：

您可以使用callbackFlow{}，并且必须在流程结束时调用awaitClose()，请参阅documentation

也许this issue link 可以帮助其他人。

【讨论】：