diff --git a/firebase-firestore/ktx/api.txt b/firebase-firestore/ktx/api.txt
index 634bb8923d9..cc32defb182 100644
--- a/firebase-firestore/ktx/api.txt
+++ b/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);
+    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);
     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);
diff --git a/firebase-firestore/ktx/ktx.gradle b/firebase-firestore/ktx/ktx.gradle
index fdc573edaa3..16338af24e5 100644
--- a/firebase-firestore/ktx/ktx.gradle
+++ b/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'
diff --git a/firebase-firestore/ktx/src/main/kotlin/com/google/firebase/firestore/ktx/Firestore.kt b/firebase-firestore/ktx/src/main/kotlin/com/google/firebase/firestore/ktx/Firestore.kt
index ec2ee225d41..5c5867856d6 100644
--- a/firebase-firestore/ktx/src/main/kotlin/com/google/firebase/firestore/ktx/Firestore.kt
+++ b/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.firestore.util.Executors.BACKGROUND_EXECUTOR
 import com.google.firebase.ktx.Firebase
 import com.google.firebase.platforminfo.LibraryVersionComponent
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
 
 /** Returns the [FirebaseFirestore] instance of the default [FirebaseApp]. */
 val Firebase.firestore: FirebaseFirestore
@@ -162,3 +171,49 @@ 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.
+ *
+ * @param metadataChanges controls metadata-only changes. Default: [MetadataChanges.EXCLUDE]
+ */
+fun DocumentReference.snapshots(
+    metadataChanges: MetadataChanges = MetadataChanges.EXCLUDE
+): Flow<DocumentSnapshot> {
+    return callbackFlow {
+        val registration = addSnapshotListener(BACKGROUND_EXECUTOR, metadataChanges) { snapshot, exception ->
+            if (exception != null) {
+                cancel(message = "Error getting DocumentReference snapshot", cause = exception)
+            } else if (snapshot != null) {
+                trySendBlocking(snapshot)
+            }
+        }
+        awaitClose { registration.remove() }
+    }
+}
+
+/**
+ * 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.
+ *
+ * @param metadataChanges controls metadata-only changes. Default: [MetadataChanges.EXCLUDE]
+ */
+fun Query.snapshots(
+    metadataChanges: MetadataChanges = MetadataChanges.EXCLUDE
+): Flow<QuerySnapshot> {
+    return callbackFlow {
+        val registration = addSnapshotListener(BACKGROUND_EXECUTOR, metadataChanges) { snapshot, exception ->
+            if (exception != null) {
+                cancel(message = "Error getting Query snapshot", cause = exception)
+            } else if (snapshot != null) {
+                trySendBlocking(snapshot)
+            }
+        }
+        awaitClose { registration.remove() }
+    }
+}