Firebase realtime snapshot listener using Coroutines

Question

I want to be able to listen to realtime updates in Firebase DB's using Kotlin coroutines in my ViewModel.

The problem is that whenever a new message is created in the collection my application freezes and won't recover from this state. I need to kill it and restart app.

For the first time it passes and I can see the previous messages on the UI. This problem happens when SnapshotListener is called for 2nd time.

My observer() function

val channel = Channel<List<MessageEntity>>()
firestore.collection(path).addSnapshotListener { data, error ->
    if (error != null) {
        channel.close(error)
    } else {
        if (data != null) {
            val messages = data.toObjects(MessageEntity::class.java)
            //till this point it gets executed^^^^
            channel.sendBlocking(messages)
        } else {
            channel.close(CancellationException("No data received"))
        }
    }
}
return channel

That's how I want to observe messages

launch(Dispatchers.IO) {
        val newMessages =
            messageRepository
                .observer()
                .receive()
    }
}

After I replacing sendBlocking() with send() I am still not getting any new messages in the channel. SnapshotListener side is executed

//channel.sendBlocking(messages) was replaced by code bellow
scope.launch(Dispatchers.IO) {
    channel.send(messages)
}
//scope is my viewModel

How to observe messages in firestore/realtime-dbs using Kotlin coroutines?

Answer 1

I have these extension functions, so I can simply get back results from the query as a Flow.

Flow is a Kotlin coroutine construct perfect for this purposes. https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/

@ExperimentalCoroutinesApi
fun CollectionReference.getQuerySnapshotFlow(): Flow<QuerySnapshot?> {
    return callbackFlow {
        val listenerRegistration =
            addSnapshotListener { querySnapshot, firebaseFirestoreException ->
                if (firebaseFirestoreException != null) {
                    cancel(
                        message = "error fetching collection data at path - $path",
                        cause = firebaseFirestoreException
                    )
                    return@addSnapshotListener
                }
                offer(querySnapshot)
            }
        awaitClose {
            Timber.d("cancelling the listener on collection at path - $path")
            listenerRegistration.remove()
        }
    }
}

@ExperimentalCoroutinesApi
fun <T> CollectionReference.getDataFlow(mapper: (QuerySnapshot?) -> T): Flow<T> {
    return getQuerySnapshotFlow()
        .map {
            return@map mapper(it)
        }
}

The following is an example of how to use the above functions.

@ExperimentalCoroutinesApi
fun getShoppingListItemsFlow(): Flow<List<ShoppingListItem>> {
    return FirebaseFirestore.getInstance()
        .collection("$COLLECTION_SHOPPING_LIST")
        .getDataFlow { querySnapshot ->
            querySnapshot?.documents?.map {
                getShoppingListItemFromSnapshot(it)
            } ?: listOf()
        }
}

// Parses the document snapshot to the desired object
fun getShoppingListItemFromSnapshot(documentSnapshot: DocumentSnapshot) : ShoppingListItem {
        return documentSnapshot.toObject(ShoppingListItem::class.java)!!
    }

And in your ViewModel class, (or your Fragment) make sure you call this from the right scope, so the listener gets removed appropriately when the user moves away from the screen.

viewModelScope.launch {
   getShoppingListItemsFlow().collect{
     // Show on the view.
   }
}