FirestoreSnapshotExt.kt

package dev.shreyaspatil.firebase.coroutines.utils

import com.google.firebase.firestore.CollectionReference
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.callbackFlow

@ExperimentalCoroutinesApi
fun CollectionReference.getSnapshotFlow() = callbackFlow {
    
    // Register listener
    val listener = addSnapshotListener { snapshot, exception ->
        
        offer(snapshot)

        // If exception occurs, cancel this scope with exception message.
        exception?.let {
            cancel(it.message.toString())
        }
    }

    awaitClose {
        // This block is executed when producer channel is cancelled
        // This function resumes with a cancellation exception.
        // Dispose listener
        listener.remove()
        cancel() 
    }
}

PostsRepository.kt

    fun getPostsRealtime() : Flow<State<List<Post>>> = mPostsCollection.getSnapshotFlow().map { snapshot ->
        if (snapshot != null && !snapshot.isEmpty) {
            State.success(snapshot.toObjects(Post::class.java))
        } else {
            State.failed<List<Post>>("No Posts!")
        }
    }