Add 'toFlow()' extensions to DocumentSnapshot and Query

firebase-firestore/firebase-firestore.gradle

@@ -144,7 +144,7 @@ dependencies {
     implementation "io.grpc:grpc-protobuf-lite:$grpcVersion"
     implementation "io.grpc:grpc-okhttp:$grpcVersion"
     implementation "io.grpc:grpc-android:$grpcVersion"
-    implementation 'com.google.android.gms:play-services-basement:18.0.0'
+    api 'com.google.android.gms:play-services-basement:18.0.0'
     implementation 'com.google.android.gms:play-services-tasks:18.0.1'
     implementation 'com.google.android.gms:play-services-base:18.0.1'
 

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> toFlow(@NonNull com.google.firebase.firestore.DocumentReference, @NonNull com.google.firebase.firestore.MetadataChanges metadataChanges = com.google.firebase.firestore.MetadataChanges.EXCLUDE);
+    method @NonNull public static kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.QuerySnapshot> toFlow(@NonNull com.google.firebase.firestore.Query, @NonNull com.google.firebase.firestore.MetadataChanges metadataChanges = com.google.firebase.firestore.MetadataChanges.EXCLUDE);
     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,7 @@ 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'
     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

@@ -14,18 +14,29 @@
 
 package com.google.firebase.firestore.ktx
 
+import android.support.multidex.BuildConfig
 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.CancellationException
+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 +173,71 @@ class FirebaseFirestoreKtxRegistrar : ComponentRegistrar {
     override fun getComponents(): List<Component<*>> =
             listOf(LibraryVersionComponent.create(LIBRARY_NAME, BuildConfig.VERSION_NAME))
 }
+
+/**
+ * Transforms a [DocumentReference] into a coroutine [Flow]
+ *
+ * **Backpressure handling**: by default this method conflates items. If the consumer isn't fast enough,
+ * it might miss some values but is always guaranteed to get the latest value. Use [bufferCapacity]
+ * 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.toFlow(
+    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)
+            }
+
+            if (snapshot != null) {
+                trySend(snapshot)
+            }
+        }
+        awaitClose { registration.remove() }
+    }
+
+    return if (bufferCapacity != null) {
+        flow.buffer(bufferCapacity)
+    } else {
+        flow
+    }
+}
+
+/**
+ * Transforms a [Query] into a coroutine [Flow]
+ *
+ * **Backpressure handling**: by default this method conflates items. If the consumer isn't fast enough,
+ * it might miss some values but is always guaranteed to get the latest value. Use [bufferCapacity]
+ * 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.toFlow(
+    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)
+            }
+
+            if (snapshot != null) {
+                trySend(snapshot)
+            }
+        }
+        awaitClose { registration.remove() }
+    }
+
+    return if (bufferCapacity != null) {
+        flow.buffer(bufferCapacity)
+    } else {
+        flow
+    }
+}