database-ktx: add callbackFlow for eventlisteners

firebase-database/ktx/ktx.gradle

@@ -50,6 +50,7 @@ dependencies {
     implementation project(':firebase-database')
     implementation 'androidx.annotation:annotation:1.1.0'
     implementation 'com.google.android.gms:play-services-tasks:18.0.1'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.4'
 
     androidTestImplementation 'junit:junit:4.12'
     androidTestImplementation "com.google.truth:truth:$googleTruthVersion"

firebase-database/ktx/src/main/kotlin/com/google/firebase/database/ktx/ChildEvent.kt

@@ -0,0 +1,43 @@
+package com.google.firebase.database.ktx
+
+import com.google.firebase.database.DataSnapshot
+
+/**
+ * Used to emit events about changes in the child locations of a given
+ * [Query] when using the [childEvents] Flow.
+ */
+sealed class ChildEvent {
+    /**
+     * Emitted when a new child is added to the location.
+     *
+     * @param snapshot An immutable snapshot of the data at the new child location
+     * @param previousChildName The key name of sibling location ordered before the new child. This
+     *     will be null for the first child node of a location.
+     */
+    data class Added(val snapshot: DataSnapshot, val previousChildName: String?) : ChildEvent()
+
+    /**
+     * Emitted when the data at a child location has changed.
+     *
+     * @param snapshot An immutable snapshot of the data at the new data at the child location
+     * @param previousChildName The key name of sibling location ordered before the child. This will
+     *     be null for the first child node of a location.
+     */
+    data class Changed(val snapshot: DataSnapshot, val previousChildName: String?) : ChildEvent()
+
+    /**
+     * Emitted when a child is removed from the location.
+     *
+     * @param snapshot An immutable snapshot of the data at the child that was removed.
+     */
+    data class Removed(val snapshot: DataSnapshot) : ChildEvent()
+
+    /**
+     * Emitted when a child location's priority changes.
+     *
+     * @param snapshot An immutable snapshot of the data at the location that moved.
+     * @param previousChildName The key name of the sibling location ordered before the child
+     *     location. This will be null if this location is ordered first.
+     */
+    data class Moved(val snapshot: DataSnapshot, val previousChildName: String?) : ChildEvent()
+}

firebase-database/ktx/src/main/kotlin/com/google/firebase/database/ktx/Database.kt

@@ -18,12 +18,20 @@ 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.database.ChildEventListener
 import com.google.firebase.database.DataSnapshot
+import com.google.firebase.database.DatabaseError
 import com.google.firebase.database.FirebaseDatabase
 import com.google.firebase.database.GenericTypeIndicator
 import com.google.firebase.database.MutableData
+import com.google.firebase.database.Query
+import com.google.firebase.database.ValueEventListener
 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.callbackFlow
 
 /** Returns the [FirebaseDatabase] instance of the default [FirebaseApp]. */
 val Firebase.database: FirebaseDatabase
@@ -59,6 +67,66 @@ inline fun <reified T> MutableData.getValue(): T? {
     return getValue(object : GenericTypeIndicator<T>() {})
 }
 
+/**
+ * Starts listening to this query and emits its values via a [Flow].
+ *
+ * - When the returned flow starts being collected, a [ValueEventListener] will be attached.
+ * - When the flow completes, the listener will be removed.
+ */
+fun Query.snapshots() = callbackFlow<DataSnapshot> {
+    val listener = addValueEventListener(object : ValueEventListener {
+        override fun onDataChange(snapshot: DataSnapshot) {
+            repo.scheduleNow {
+                trySendBlocking(snapshot)
+            }
+        }
+
+        override fun onCancelled(error: DatabaseError) {
+            cancel(message = "Error getting Query snapshot", cause = error.toException())
+        }
+    })
+    awaitClose { removeEventListener(listener) }
+}
+
+/**
+ * Starts listening to this query's child events and emits its values via a [Flow].
+ *
+ * - When the returned flow starts being collected, a [ChildEventListener] will be attached.
+ * - When the flow completes, the listener will be removed.
+ */
+fun Query.childEvents() = callbackFlow<ChildEvent> {
+    val listener = addChildEventListener(object : ChildEventListener {
+        override fun onChildAdded(snapshot: DataSnapshot, previousChildName: String?) {
+            repo.scheduleNow {
+                trySendBlocking(ChildEvent.Added(snapshot, previousChildName))
+            }
+        }
+
+        override fun onChildChanged(snapshot: DataSnapshot, previousChildName: String?) {
+            repo.scheduleNow {
+                trySendBlocking(ChildEvent.Changed(snapshot, previousChildName))
+            }
+        }
+
+        override fun onChildRemoved(snapshot: DataSnapshot) {
+            repo.scheduleNow {
+                trySendBlocking(ChildEvent.Removed(snapshot))
+            }
+        }
+
+        override fun onChildMoved(snapshot: DataSnapshot, previousChildName: String?) {
+            repo.scheduleNow {
+                trySendBlocking(ChildEvent.Moved(snapshot, previousChildName))
+            }
+        }
+
+        override fun onCancelled(error: DatabaseError) {
+            cancel(message = "Error getting Query childEvent", cause = error.toException())
+        }
+    })
+    awaitClose { removeEventListener(listener) }
+}
+
 internal const val LIBRARY_NAME: String = "fire-db-ktx"
 
 /** @suppress */