Kotlin Firestore API (Collection documentChanges) to Flow

test.kt

@ExperimentalCoroutinesApi
fun CollectionReference.toFlow(): Flow<DocumentChange> = callbackFlow {
    val snapshotListener = this@toFlow.addSnapshotListener { snapshots, e ->
        if (e != null) {
            logger.error { "${this@toFlow.path} listen:error:$e" }
            cancel(CancellationException("Collection Listener Error in ${this@toFlow.path}", e))
            return@addSnapshotListener
        }
        for (dc in snapshots!!.documentChanges) {
            offer(dc)
        }
    }
    awaitClose {
        logger.info { "Closing collection's flow, removing listener." }
        snapshotListener.remove()
    }
}

fun main() {
  val myDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
  runBlocking(myDispatcher) {
      commandsRef.toFlow()
          .filter { dc -> dc.type == DocumentChange.Type.ADDED }
          .collect { dc ->
              println(dc.document["somepropkey"] as String)
          }
  }
}