Realtime API

Author: tom-andersen

firebase-firestore/src/main/java/com/google/firebase/firestore/ListenerRegistration.java

@@ -14,6 +14,10 @@
 
 package com.google.firebase.firestore;
 
+import android.app.Activity;
+import androidx.annotation.NonNull;
+import com.google.firebase.firestore.core.ActivityScope;
+
 /** Represents a listener that can be removed by calling {@link #remove()}. */
 public interface ListenerRegistration {
 
@@ -22,4 +26,13 @@ public interface ListenerRegistration {
    * call, subsequent calls have no effect.
    */
   void remove();
+
+  /**
+   * The listener will be automatically removed during {@link Activity#onStop}.
+   *
+   * @param activity The activity to scope the listener to.
+   */
+  default void scope(@NonNull Activity activity) {
+    ActivityScope.bind(activity, this);
+  }
 }

firebase-firestore/src/main/java/com/google/firebase/firestore/Pipeline.kt

@@ -52,9 +52,12 @@ import com.google.firebase.firestore.pipeline.Stage
 import com.google.firebase.firestore.pipeline.UnionStage
 import com.google.firebase.firestore.pipeline.UnnestStage
 import com.google.firebase.firestore.pipeline.WhereStage
+import com.google.firebase.firestore.util.Executors
 import com.google.firestore.v1.ExecutePipelineRequest
 import com.google.firestore.v1.StructuredPipeline
 import com.google.firestore.v1.Value
+import java.util.concurrent.Executor
+import kotlinx.coroutines.flow.Flow
 
 open class AbstractPipeline
 internal constructor(
@@ -775,6 +778,38 @@ internal constructor(
 
   fun execute(options: RealtimePipelineOptions): Task<PipelineSnapshot> = execute(options.options)
 
+  fun snapshots(): Flow<RealtimePipelineSnapshot> = snapshots(RealtimePipelineOptions.DEFAULT)
+
+  fun snapshots(options: RealtimePipelineOptions): Flow<RealtimePipelineSnapshot> {
+    throw NotImplementedError()
+  }
+
+  fun addSnapshotListener(listener: EventListener<RealtimePipelineSnapshot>): ListenerRegistration =
+    addSnapshotListener(
+      Executors.DEFAULT_CALLBACK_EXECUTOR,
+      RealtimePipelineOptions.DEFAULT,
+      listener
+    )
+
+  fun addSnapshotListener(
+    options: RealtimePipelineOptions,
+    listener: EventListener<RealtimePipelineSnapshot>
+  ): ListenerRegistration =
+    addSnapshotListener(Executors.DEFAULT_CALLBACK_EXECUTOR, options, listener)
+
+  fun addSnapshotListener(
+    executor: Executor,
+    listener: EventListener<RealtimePipelineSnapshot>
+  ): ListenerRegistration = addSnapshotListener(executor, RealtimePipelineOptions.DEFAULT, listener)
+
+  fun addSnapshotListener(
+    executor: Executor,
+    options: RealtimePipelineOptions,
+    listener: EventListener<RealtimePipelineSnapshot>
+  ): ListenerRegistration {
+    throw NotImplementedError()
+  }
+
   fun limit(limit: Int): RealtimePipeline = append(LimitStage(limit))
 
   fun offset(offset: Int): RealtimePipeline = append(OffsetStage(offset))
@@ -818,8 +853,6 @@ internal constructor(
   }
 }
 
-/**
- */
 class PipelineSnapshot
 internal constructor(executionTime: Timestamp, results: List<PipelineResult>) :
   Iterable<PipelineResult> {
@@ -833,6 +866,25 @@ internal constructor(executionTime: Timestamp, results: List<PipelineResult>) :
   override fun iterator() = results.iterator()
 }
 
+class RealtimePipelineSnapshot
+internal constructor(
+  executionTime: Timestamp?,
+  metadata: SnapshotMetadata,
+  results: List<PipelineResult>
+) : Iterable<PipelineResult> {
+
+  /** The time at which the pipeline producing this result is executed. */
+  val executionTime: Timestamp? = executionTime
+
+  /** Metadata about this snapshot, concerning its source and if it has local modifications. */
+  val metadata: SnapshotMetadata = metadata
+
+  /** List of all the results */
+  val results: List<PipelineResult> = results
+
+  override fun iterator() = results.iterator()
+}
+
 class PipelineResult
 internal constructor(
   private val firestore: FirebaseFirestore,

firebase-firestore/src/main/java/com/google/firebase/firestore/pipeline/options.kt

@@ -15,6 +15,8 @@
 package com.google.firebase.firestore.pipeline
 
 import com.google.common.collect.ImmutableMap
+import com.google.firebase.firestore.DocumentSnapshot
+import com.google.firebase.firestore.MetadataChanges
 import com.google.firebase.firestore.model.Values
 import com.google.firestore.v1.ArrayValue
 import com.google.firestore.v1.MapValue
@@ -156,8 +158,31 @@ class PipelineOptions private constructor(options: InternalOptions) :
   fun withIndexMode(indexMode: IndexMode): PipelineOptions = with("index_mode", indexMode.value)
 }
 
-class RealtimePipelineOptions private constructor(options: InternalOptions) :
-  AbstractOptions<RealtimePipelineOptions>(options) {
+class RealtimePipelineOptions
+private constructor(
+  internal val offlineServerTimestamps: DocumentSnapshot.ServerTimestampBehavior,
+  internal val metadataChanges: MetadataChanges,
+  options: InternalOptions
+) : AbstractOptions<RealtimePipelineOptions>(options) {
 
-  override fun self(options: InternalOptions) = RealtimePipelineOptions(options)
+  override fun self(options: InternalOptions) =
+    RealtimePipelineOptions(offlineServerTimestamps, metadataChanges, options)
+
+  companion object {
+    @JvmField
+    val DEFAULT: RealtimePipelineOptions =
+      RealtimePipelineOptions(
+        DocumentSnapshot.ServerTimestampBehavior.NONE,
+        MetadataChanges.EXCLUDE,
+        InternalOptions.EMPTY
+      )
+  }
+
+  fun withOfflineServerTimestamps(
+    offlineServerTimestamps: DocumentSnapshot.ServerTimestampBehavior
+  ): RealtimePipelineOptions =
+    RealtimePipelineOptions(offlineServerTimestamps, metadataChanges, options)
+
+  fun withMetadataChanges(metadataChanges: MetadataChanges): RealtimePipelineOptions =
+    RealtimePipelineOptions(offlineServerTimestamps, metadataChanges, options)
 }