wrapping callbacks that fire more than once about kotlinx.coroutines

[beyondeye]

ok so I find the solution: I should have used suspendCancellableCoroutine

[beyondeye]

suspendCancellableCoroutine build a SuspendableCoroutine that exposes (among other stuff) the two methods

public fun tryResume(value: T, idempotent: Any? = null): Any?
public fun tryResumeWithException(exception: Throwable): Any?

that basically do what I want: they check the continuation state (that I cannot access directly because it is internal data), if it was already resumed.
About Firebase: for that specific API (getFileDownloadUrl) the callback is called twice almost always. There are also other Firebase APIs that behave the same way. I don't know the cause. It is a documented behavior, but I never actually understood why this should happen at all. But this is how it works.

[fvasco]

After first "complete event" the completion listener is unuseful therefore you should remove it.

IMHO this isn't a Continuation issue.

[beyondeye]

Unfortunately I could not find a way to remove the listener.

[elizarov]

I'm curios what difference suspendCancellableCoroutine makes for you and what is exactly the story with firebase invoking callback more than once? In what scenarios can this happen?

[elizarov]

Thanks. Can you point me to the place where Firebase documents multiple callback invocations, please? I want to understand the root cause of it.

[beyondeye]

for example here

[adam-hurwitz]

@elizarov, This StackOverflow outlining Firebase realtime snapshot listener using Coroutines is a great example.

I've used the following pattern.

Realtime updates

The extension function awaitRealtime has checks including verifying the state of the continuation in order to see whether it is in isActive state. This is important because the function is called when the user's main feed of content is updated either by a lifecycle event, refreshing the feed manually, or removing content from their feed. Without this check there will be a crash.

ExtenstionFuction.kt

suspend fun Query.awaitRealtime() = suspendCancellableCoroutine<QuerySnapshot?> { cont ->
    addSnapshotListener({ value, error ->
        if (error == null && cont.isActive && !value!!.isEmpty) {
            cont.resume(value)
        }
    })
}

In order to handle errors the try/catch pattern is used. In a future version the Repository LiveData will be refactored to use the Flow pattern. The ViewModel will observe the Flows and save data to the UI using the existing LiveData states.

Repository.kt

object ContentRepository {
    fun getMainFeedList(scope: CoroutineScope, isRealtime: Boolean, timeframe: Timestamp) =
            liveData<Lce<PagedListResult>>(scope.coroutineContext) {
                val lce = this
                lce.emit(Loading())
                val labeledSet = HashSet<String>()
                val user = getUesrCall(...)
                getLabeledContent(user, timeframe, labeledSet, SAVE_COLLECTION, lce)
                getLoggedInNonRealtimeContent(timeframe, labeledSet, lce) 
            }

    // Realtime updates with 'awaitRealtime' used
    private suspend fun getLabeledContent(user: CollectionReference, timeframe: Timestamp,
                                          labeledSet: HashSet<String>,
                                          saveCollection: String,
                                          lce: LiveDataScope<Lce<PagedListResult>>) =
            try {
                val list = ArrayList<Content?>()
                user.document(COLLECTIONS_DOCUMENT)
                        .collection(saveCollection)
                        .orderBy(TIMESTAMP, DESCENDING)
                        .whereGreaterThanOrEqualTo(TIMESTAMP, timeframe)
                        .awaitRealtime()?.documentChanges?.map { doc ->
                    doc.document.toObject(Content::class.java).let { content ->
                        list.add(content)
                        labeledSet.add(content.id)
                    }
                }
                database.contentDao().insertContentList(list)
            } catch (error: FirebaseFirestoreException) {
                lce.emit(Error(PagedListResult(null,
                        "Error retrieving user save_collection: " +
                                "${error.localizedMessage}")))
            }

    // One time updates with 'await' used
    private suspend fun getLoggedInNonRealtimeContent(timeframe: Timestamp,
                                                      labeledSet: HashSet<String>,
                                                      lce: LiveDataScope<Lce<PagedListResult>>) {
            try {
                val list = ArrayList<Content?>()
                contentEnCollection.orderBy(TIMESTAMP, DESCENDING)
                    .whereGreaterThanOrEqualTo(TIMESTAMP, timeframe).get().await()
                    .documentChanges
                    ?.map { change -> change.document.toObject(Content::class.java) }
                    ?.filter { content -> !labeledSet.contains(content.id) }
                    ?.map { content -> list.add(content) }
                database.contentDao().insertContentList(list)
                lce.emit(Lce.Content(PagedListResult(
                    queryMainContentList(timeframe), "")))
            } catch (error: FirebaseFirestoreException) {
                lce.emit(Error(PagedListResult(
                    null,
                    CONTENT_LOGGED_IN_NON_REALTIME_ERROR +
                            "${error.localizedMessage}")))
            }
        }
}