Still a flow/coroutine noob. I'm using firebase fi...

# coroutines

[Colton Idle]

Still a flow/coroutine noob. I'm using firebase firestore and it has a listener approach to listen to updates on my data. I want to convert this into a Flowable so I can observe it.

Is this an okay approach

suspend fun getThingFlow(myId: String): Flow<Thing?> {
    val thingFlow = MutableStateFlow<Thing?>(null)
    FirebaseFirestore.getInstance()
        .document("things/$myId")
        .addSnapshotListener { value, error ->
            thingFlow.emit(value!!.toObject<Thing>())
        }

    return thingFlow
}

[Colton Idle]

It also seems like

suspendCoroutine {

and

callbackFlow

are popular. Just overall confused on which would be the best approach here.

[KamilH]

There is a special API for transforming callback-based API to coroutines:

suspend fun getThingFlow(myId: String): Flow<Thing?> =
        suspendCancellableCoroutine<Thing> { continuation ->
            FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    continuation.resume(value!!.toObject<Thing>()) // note that this `resume` method is extension function one, so you need to import it
                }
            it.invokeOnCancellation {
                // do the cleanup here
            }
        }

[KamilH]

I don’t know FirebaseFirestore API, so be careful copy-pasting it, but generally it should be something like above

[Colton Idle]

@KamilH weird that the api is "coroutine" (suspendCancellableCoroutine) and not "flow" (like suspendCancellableFlow)

Makes it seem like I wont keep getting updates besides the first one. But okay. I will give it a whirl. thanks

[Colton Idle]

Oh yeah. Docs for suspendCancellableCoroutine say "For multi-shot callback APIs see callbackFlow."

[KamilH]

Yes, sorry, you’re right I miss-read your request. You should use `callbackFlow`:

suspend fun getThingFlow(myId: String): Flow<Thing?> =
        callbackFlow<Thing?> {
            FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    trySend(value!!.toObject<Thing>()) 
                }
            
            invokeOnClose {
                // do the cleanup here
            }
        }

[KamilH]

As I said, I don’t know FirebaseFirestore API, so if there is any indication that this stream is finished (closed) you can use close() function to close this flow

[Colton Idle]

Cool. I will keep researching. This gets me closer seemingly... 🤞

[Colton Idle]

oh fuck yeah. it works! LIVE UPDATES!

Thanks for teaching!

[Tgo1014]

callbackFlow is the way to go, but honestly I’m impressed there’s no flow api for Firebase SDK yet

[hfhbd]

BTW StateFlow is a hot flow, callbackFlow is a cold one.

[Joffrey]

+1 to @hfhbd, with the callbackFlow approach, the function shouldn't be suspending. Also, the listener should be assigned to a variable so it can be unregistered in awaitClose. (see the docs for callbackFlow, they are pretty helpful)

[Colton Idle]

With @Joffreys advice I changed it to:

fun getThingFlow(myId: String): Flow<Thing?> =
        callbackFlow<Thing?> {
            val listener = FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    trySend(value!!.toObject<Thing>()) 
                }
            
            invokeOnClose {
                listener.close()
            }
        }

Thanks everyone!

[hfhbd]

@Colton Idle Not exactly 😄

fun getThingFlow(myId: String): Flow<Thing?> =
        callbackFlow<Thing?> {
            val listener = FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    trySend(value!!.toObject<Thing>()) 
                }
            
            invokeOnClose {
               close()
            }
        awaitClose {
          listener.close()
        }

[Colton Idle]

ooOoh. invokeOnClose and awaitClose?

[Joffrey]

I don't think invokeOnClose should be called at all here. Only awaitClose

[hfhbd]

@Joffrey Yeah, the invokeOnClose function should be part of listener to finish/close the flow when all items are emitted. Otherwise, this flow will never finish by itself.

[Joffrey]

No, invokeOnClose shouldn't be called in the listener either, only close() should (when you want to close the channel of the flow). invokeOnClose is a callback on the SendChannel itself, why doesn't really have its place in a callbackFlow setup AFAIU

[hfhbd]

I mean this:

fun getThingFlow(myId: String): Flow<Thing?> =
        callbackFlow<Thing?> {
            val listener = FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    trySend(value!!.toObject<Thing>()) 
                }
            listener.onError { // I dont know the FireStore API
close(it)
}
            listener.invokeOnClose { // or onFinished. I dont know the FireStore API...
               close()
            }
        awaitClose {
          listener.close()
        }

[Joffrey]

Oh ok, in your previous snippet invokeOnClose was that of the SendChannel, and is an existing function. I didn't get that you were talking about a hypothetical closing callback of the listener. There is no such thing AFAIK in this case, it's just a subscription to receive updates. It has no end in itself, so the only way to end the flow is by cancelling it from the collector side, which awaitClose is meant to handle:

fun getThingFlow(myId: String): Flow<Thing?> =
        callbackFlow<Thing?> {
            val listener = FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    trySend(value!!.toObject<Thing>()) 
                }
            
            awaitClose {
                listener.close()
            }
        }

[hfhbd]

Thanks, I wasn't aware this is an infinite stream.

[Colton Idle]

Wait, so @Joffrey's last example is what I should go with? 😅

[Joffrey]

Yes, I believe so

[Colton Idle]

i also just submitted a FR to firebase.

[Joffrey]

If you don't call awaitClose anywhere, your flow will terminate immediately. invokeOnClose just registers a handler but doesn't wait

[Joffrey]

Please link your issue here, it might help other people 😉

[Colton Idle]

theres no public issue. the FR for firebase is only through a form

[Joffrey]

Ah ok, fair enough then 😄

[Colton Idle]

Okay. Here's my final usage

fun getThingFlow(myId: String): Flow<Thing?> =
        callbackFlow<Thing?> {
            val listener = FirebaseFirestore.getInstance()
                .document("things/$myId")
                .addSnapshotListener { value, error ->
                    trySend(value!!.toObject<Thing>()) 
                }
            
            awaitClose {
                listener.remove()
            }
        }

The only real change I made was listener.remove() since there is no close() api (i dont know where I got that from)

[Colton Idle]

For anyone following along. It looks like firebase is getting flow support!
https://github.com/firebase/firebase-android-sdk/pull/1252#issuecomment-1213179027