Add 'toFlow()' extensions to DocumentSnapshot and Query

firebase-firestore/ktx/api.txt

@@ -10,6 +10,8 @@ package com.google.firebase.firestore.ktx {
     method @Nullable public static inline <reified T> T getField(@NonNull com.google.firebase.firestore.DocumentSnapshot, @NonNull com.google.firebase.firestore.FieldPath fieldPath);
     method @Nullable public static inline <reified T> T getField(@NonNull com.google.firebase.firestore.DocumentSnapshot, @NonNull com.google.firebase.firestore.FieldPath fieldPath, @NonNull com.google.firebase.firestore.DocumentSnapshot.ServerTimestampBehavior serverTimestampBehavior);
     method @NonNull public static com.google.firebase.firestore.FirebaseFirestore getFirestore(@NonNull com.google.firebase.ktx.Firebase);
+    method @NonNull public static kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.DocumentSnapshot> snapshots(@NonNull com.google.firebase.firestore.DocumentReference, @NonNull com.google.firebase.firestore.MetadataChanges metadataChanges = com.google.firebase.firestore.MetadataChanges.EXCLUDE, @Nullable Integer bufferCapacity = -1);
+    method @NonNull public static kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.QuerySnapshot> snapshots(@NonNull com.google.firebase.firestore.Query, @NonNull com.google.firebase.firestore.MetadataChanges metadataChanges = com.google.firebase.firestore.MetadataChanges.EXCLUDE, @Nullable Integer bufferCapacity = -1);
     method @Nullable public static inline <reified T> T toObject(@NonNull com.google.firebase.firestore.DocumentSnapshot);
     method @Nullable public static inline <reified T> T toObject(@NonNull com.google.firebase.firestore.DocumentSnapshot, @NonNull com.google.firebase.firestore.DocumentSnapshot.ServerTimestampBehavior serverTimestampBehavior);
     method @NonNull public static inline <reified T> T toObject(@NonNull com.google.firebase.firestore.QueryDocumentSnapshot);

firebase-firestore/ktx/ktx.gradle

@@ -55,6 +55,8 @@ dependencies {
     implementation project(':firebase-common:ktx')
     implementation project(':firebase-firestore')
     implementation 'androidx.annotation:annotation:1.1.0'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.2'
+    implementation 'com.google.android.gms:play-services-basement:18.0.0'
     testImplementation project(':firebase-database-collection')
     testImplementation 'org.mockito:mockito-core:2.25.0'
     testImplementation 'com.fasterxml.jackson.core:jackson-databind:2.9.8'

firebase-firestore/ktx/src/main/kotlin/com/google/firebase/firestore/ktx/Firestore.kt

@@ -18,14 +18,23 @@ import androidx.annotation.Keep
 import com.google.firebase.FirebaseApp
 import com.google.firebase.components.Component
 import com.google.firebase.components.ComponentRegistrar
+import com.google.firebase.firestore.DocumentReference
 import com.google.firebase.firestore.DocumentSnapshot
 import com.google.firebase.firestore.FieldPath
 import com.google.firebase.firestore.FirebaseFirestore
 import com.google.firebase.firestore.FirebaseFirestoreSettings
+import com.google.firebase.firestore.MetadataChanges
+import com.google.firebase.firestore.Query
 import com.google.firebase.firestore.QueryDocumentSnapshot
 import com.google.firebase.firestore.QuerySnapshot
 import com.google.firebase.ktx.Firebase
 import com.google.firebase.platforminfo.LibraryVersionComponent
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.callbackFlow
 
 /** Returns the [FirebaseFirestore] instance of the default [FirebaseApp]. */
 val Firebase.firestore: FirebaseFirestore
@@ -162,3 +171,73 @@ class FirebaseFirestoreKtxRegistrar : ComponentRegistrar {
     override fun getComponents(): List<Component<*>> =
             listOf(LibraryVersionComponent.create(LIBRARY_NAME, BuildConfig.VERSION_NAME))
 }
+
+/**
+ * Starts listening to the document referenced by this `DocumentReference` with the given options and emits its values via a [Flow].
+ *
+ * - When the returned flow starts being collected, an [EventListener] will be attached.
+ * - When the flow completes, the listener will be removed.
+ *
+ * Backpressure: by default the returned flow is conflated. If the consumer isn't fast enough,
+ * it might miss some values but is always guaranteed to get the latest value. Use the [bufferCapacity] parameter
+ * to change that behaviour.
+ *
+ * @param metadataChanges controls metadata-only changes. Default: [MetadataChanges.EXCLUDE]
+ * @param bufferCapacity the buffer capacity as in [Flow.buffer] or null to not buffer at all
+ */
+fun DocumentReference.snapshots(
+    metadataChanges: MetadataChanges = MetadataChanges.EXCLUDE,
+    bufferCapacity: Int? = Channel.CONFLATED
+): Flow<DocumentSnapshot> {
+    val flow = callbackFlow {
+        val registration = addSnapshotListener(metadataChanges) { snapshot, exception ->
+            if (exception != null) {
+                cancel(message = "Error getting DocumentReference snapshot", cause = exception)
+            } else if (snapshot != null) {
+                trySend(snapshot)
+            }
+        }
+        awaitClose { registration.remove() }
+    }
+
+    return if (bufferCapacity != null) {
+        flow.buffer(bufferCapacity)
+    } else {
+        flow
+    }
+}
+
+/**
+ * Starts listening to this query with the given options and emits its values via a [Flow].
+ *
+ * - When the returned flow starts being collected, an [EventListener] will be attached.
+ * - When the flow completes, the listener will be removed.
+ *
+ * Backpressure: by default the returned flow is conflated. If the consumer isn't fast enough,
+ * it might miss some values but is always guaranteed to get the latest value. Use the [bufferCapacity] parameter
+ * to change that behaviour.
+ *
+ * @param metadataChanges controls metadata-only changes. Default: [MetadataChanges.EXCLUDE]
+ * @param bufferCapacity the buffer capacity as in [Flow.buffer] or null to not buffer at all
+ */
+fun Query.snapshots(
+    metadataChanges: MetadataChanges = MetadataChanges.EXCLUDE,
+    bufferCapacity: Int? = Channel.CONFLATED
+): Flow<QuerySnapshot> {
+    val flow = callbackFlow {
+        val registration = addSnapshotListener(metadataChanges) { snapshot, exception ->
+            if (exception != null) {
+                cancel(message = "Error getting Query snapshot", cause = exception)
+            } else if (snapshot != null) {
+                trySend(snapshot)
+            }
+        }
+        awaitClose { registration.remove() }
+    }
+
+    return if (bufferCapacity != null) {
+        flow.buffer(bufferCapacity)
+    } else {
+        flow
+    }
+}