Emit State from Firebase listener - Kotlin Flow

user_13542359 0
Viewed 2814

I want to use Kotlin Flow to handle the FirebaseAuth State. I know that code below is wrong but I jave no idea how to fix it. I tried with channelFlow and it crashed always when I want send or 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))
            }
        }
}

}

the coroutine inside Listener throws

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'

the error when I use send()

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

The flow ends when the block of code runs to completion.

Instead of using a callback, you can use the integration to make it a suspending function.

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))
    }
}
user_53630110

Method 1 (Recommended)

Add this dependency to your app's build.gradle file:

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

This dependency gives you access to two extension functions:

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

This approach helps eliminate boilerplate code:

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

Method 2

As stated in the error, you should use channelFlow instead. At the time of writing, it's still experimental but works well.

Example using firebase's addSuccessListener and 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()
    }
user_138348950

You can not change the context within a flow builder. What you can do instead is when you call your method registerFlow() is to use flowOn() and provide the correct context of execution. In your case you will have something like this:

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