#3 Remove remaining references of RxFirestore from remote storage

ground/build.gradle

@@ -186,7 +186,6 @@ dependencies {
     implementation 'com.google.firebase:firebase-messaging'
     implementation 'com.google.firebase:firebase-messaging-directboot'
     implementation 'com.google.firebase:firebase-messaging-ktx'
-    implementation 'com.github.FrangSierra:RxFirebase:1.5.7'
 
     // Hilt
     implementation "com.google.dagger:hilt-android:$project.hiltVersion"

ground/src/main/java/com/google/android/ground/persistence/remote/RemoteDataStore.kt

@@ -21,8 +21,6 @@ import com.google.android.ground.model.User
 import com.google.android.ground.model.locationofinterest.LocationOfInterest
 import com.google.android.ground.model.mutation.Mutation
 import com.google.android.ground.model.submission.Submission
-import com.google.android.ground.rx.annotations.Cold
-import io.reactivex.Flowable
 
 /**
  * Defines API for accessing data in a remote data store. Implementations must ensure all
@@ -43,14 +41,6 @@ interface RemoteDataStore {
    */
   suspend fun loadTermsOfService(): TermsOfService?
 
-  /**
-   * Returns all LOIs in the specified survey, then continues to emit any remote updates to the set
-   * of LOIs in the survey until all subscribers have been disposed.
-   */
-  fun loadLocationsOfInterestOnceAndStreamChanges(
-    survey: Survey
-  ): @Cold(stateful = true, terminates = false) Flowable<RemoteDataEvent<LocationOfInterest>>
-
   /** Returns all LOIs in the specified survey. Main-safe. */
   suspend fun loadLocationsOfInterest(survey: Survey): List<LocationOfInterest>
 

ground/src/main/java/com/google/android/ground/persistence/remote/firebase/FirestoreDataStore.kt

@@ -25,18 +25,14 @@ import com.google.android.ground.model.mutation.Mutation
 import com.google.android.ground.model.mutation.SubmissionMutation
 import com.google.android.ground.model.submission.Submission
 import com.google.android.ground.persistence.remote.DataStoreException
-import com.google.android.ground.persistence.remote.RemoteDataEvent
 import com.google.android.ground.persistence.remote.RemoteDataStore
 import com.google.android.ground.persistence.remote.firebase.schema.GroundFirestore
-import com.google.android.ground.rx.Schedulers
-import com.google.android.ground.rx.annotations.Cold
 import com.google.android.ground.system.ApplicationErrorManager
 import com.google.firebase.crashlytics.FirebaseCrashlytics
 import com.google.firebase.firestore.WriteBatch
 import com.google.firebase.functions.FirebaseFunctions
 import com.google.firebase.ktx.Firebase
 import com.google.firebase.messaging.ktx.messaging
-import io.reactivex.Flowable
 import javax.inject.Inject
 import javax.inject.Singleton
 import kotlinx.coroutines.CoroutineDispatcher
@@ -54,7 +50,6 @@ internal constructor(
   private val errorManager: ApplicationErrorManager,
   @IoDispatcher private val ioDispatcher: CoroutineDispatcher,
   val db: GroundFirestore,
-  val schedulers: Schedulers
 ) : RemoteDataStore {
 
   /**
@@ -87,19 +82,6 @@ internal constructor(
   override suspend fun loadSurveySummaries(user: User): List<Survey> =
     withContext(ioDispatcher) { db.surveys().getReadable(user) }
 
-  override fun loadLocationsOfInterestOnceAndStreamChanges(
-    survey: Survey
-  ): @Cold(stateful = true, terminates = false) Flowable<RemoteDataEvent<LocationOfInterest>> =
-    db
-      .surveys()
-      .survey(survey.id)
-      .lois()
-      .loadOnceAndStreamChanges(survey)
-      .onErrorResumeNext { e: Throwable ->
-        if (shouldInterceptException(e)) Flowable.never() else Flowable.error(e)
-      }
-      .subscribeOn(schedulers.io())
-
   override suspend fun loadLocationsOfInterest(survey: Survey) =
     db.surveys().survey(survey.id).lois().locationsOfInterest(survey)
 

ground/src/main/java/com/google/android/ground/persistence/remote/firebase/base/FluentCollectionReference.kt

@@ -32,12 +32,7 @@ protected constructor(
   protected val defaultDispatcher: CoroutineDispatcher = Dispatchers.Default
 ) {
 
-  /**
-   * Returns a Completable that completes immediately on subscribe if network is available, or fails
-   * in error if not.
-   */
-  private fun requireActiveNetwork() =
-    requireActiveNetwork(reference.firestore.app.applicationContext)
+  private val context = reference.firestore.app.applicationContext
 
   /**
    * Runs the specified query, returning a Single containing a List of values created by applying
@@ -48,7 +43,7 @@ protected constructor(
     query: Query,
     mappingFunction: Function<DocumentSnapshot, T>
   ): List<T> {
-    requireActiveNetwork()
+    requireActiveNetwork(context)
     val querySnapshot = query.get().await()
     return querySnapshot.documents.filter { it.exists() }.map { mappingFunction.apply(it) }
   }

ground/src/main/java/com/google/android/ground/persistence/remote/firebase/base/FluentFirestore.kt

@@ -16,14 +16,8 @@
 
 package com.google.android.ground.persistence.remote.firebase.base
 
-import com.google.android.ground.rx.annotations.Cold
-import com.google.firebase.firestore.DocumentSnapshot
 import com.google.firebase.firestore.FirebaseFirestore
-import com.google.firebase.firestore.QuerySnapshot
 import com.google.firebase.firestore.WriteBatch
-import io.reactivex.Maybe
-import io.reactivex.Single
-import java8.util.function.Function
 
 /** Base class for representing Firestore databases as object hierarchies. */
 open class FluentFirestore protected constructor(private val db: FirebaseFirestore) {
@@ -32,20 +26,4 @@ open class FluentFirestore protected constructor(private val db: FirebaseFiresto
 
   // TODO: Wrap in fluent version of WriteBatch.
   fun batch(): WriteBatch = db.batch()
-
-  companion object {
-    /**
-     * Applies the provided mapping function to each document in the specified query snapshot, if
-     * present. If no results are present, completes with an empty list.
-     */
-    fun <T> toSingleList(
-      result: Maybe<QuerySnapshot>,
-      mappingFunction: Function<DocumentSnapshot, T>
-    ): @Cold Single<List<T>> =
-      result
-        .map { querySnapshot: QuerySnapshot ->
-          querySnapshot.documents.map { mappingFunction.apply(it) }
-        }
-        .toSingle(emptyList())
-  }
 }

ground/src/main/java/com/google/android/ground/persistence/remote/firebase/schema/LoiCollectionReference.kt

@@ -18,30 +18,17 @@ package com.google.android.ground.persistence.remote.firebase.schema
 
 import com.google.android.ground.model.Survey
 import com.google.android.ground.model.locationofinterest.LocationOfInterest
-import com.google.android.ground.persistence.remote.RemoteDataEvent
 import com.google.android.ground.persistence.remote.firebase.base.FluentCollectionReference
 import com.google.android.ground.persistence.remote.firebase.schema.LoiConverter.toLoi
-import com.google.android.ground.rx.annotations.Cold
 import com.google.firebase.firestore.CollectionReference
-import com.google.firebase.firestore.DocumentSnapshot
 import com.google.firebase.firestore.QuerySnapshot
-import durdinapps.rxfirebase2.RxFirestore
-import io.reactivex.Flowable
 import kotlinx.coroutines.tasks.await
 import kotlinx.coroutines.withContext
 import timber.log.Timber
 
 class LoiCollectionReference internal constructor(ref: CollectionReference) :
   FluentCollectionReference(ref) {
 
-  /** Retrieves all lois in the survey, then streams changes to the remote db incrementally. */
-  fun loadOnceAndStreamChanges(
-    survey: Survey
-  ): @Cold(terminates = false) Flowable<RemoteDataEvent<LocationOfInterest>> =
-    RxFirestore.observeQueryRef(reference()).flatMapIterable { snapshot: QuerySnapshot ->
-      toRemoteDataEvents(survey, snapshot)
-    }
-
   fun loi(id: String) = LoiDocumentReference(reference().document(id))
 
   /** Retrieves all LOIs in the specified survey. Main-safe. */
@@ -55,10 +42,4 @@ class LoiCollectionReference internal constructor(ref: CollectionReference) :
         // Filter out bad results and log.
         .mapNotNull { it.onFailure { t -> Timber.e(t) }.getOrNull() }
     }
-
-  private fun toRemoteDataEvents(
-    survey: Survey,
-    snapshot: QuerySnapshot
-  ): Iterable<RemoteDataEvent<LocationOfInterest>> =
-    QuerySnapshotConverter.toEvents(snapshot) { doc: DocumentSnapshot -> toLoi(survey, doc) }
 }

ground/src/main/java/com/google/android/ground/persistence/remote/firebase/schema/SubmissionDocumentReference.kt

@@ -17,24 +17,14 @@
 package com.google.android.ground.persistence.remote.firebase.schema
 
 import com.google.android.ground.model.User
-import com.google.android.ground.model.locationofinterest.LocationOfInterest
 import com.google.android.ground.model.mutation.Mutation
 import com.google.android.ground.model.mutation.SubmissionMutation
-import com.google.android.ground.model.submission.Submission
 import com.google.android.ground.persistence.remote.firebase.base.FluentDocumentReference
-import com.google.android.ground.rx.annotations.Cold
 import com.google.firebase.firestore.DocumentReference
-import com.google.firebase.firestore.DocumentSnapshot
 import com.google.firebase.firestore.WriteBatch
-import durdinapps.rxfirebase2.RxFirestore
-import io.reactivex.Maybe
 
 class SubmissionDocumentReference internal constructor(ref: DocumentReference) :
   FluentDocumentReference(ref) {
-  operator fun get(locationOfInterest: LocationOfInterest): @Cold Maybe<Submission> =
-    RxFirestore.getDocument(reference()).map { doc: DocumentSnapshot ->
-      SubmissionConverter.toSubmission(locationOfInterest, doc)
-    }
 
   /** Appends the operation described by the specified mutation to the provided write batch. */
   fun addMutationToBatch(mutation: SubmissionMutation, user: User, batch: WriteBatch) {

ground/src/main/java/com/google/android/ground/system/NetworkManager.kt

@@ -31,10 +31,7 @@ object NetworkManager {
     return networkInfo?.isConnected ?: false
   }
 
-  /**
-   * Returns a Completable that completes immediately on subscribe if network is available, or fails
-   * in error if not.
-   */
+  /** Throws an error if network isn't available. */
   fun requireActiveNetwork(context: Context) {
     if (!isNetworkAvailable(context)) {
       throw ConnectException()

sharedTest/src/main/kotlin/com/sharedtest/persistence/remote/FakeRemoteDataStore.kt

@@ -21,9 +21,7 @@ import com.google.android.ground.model.User
 import com.google.android.ground.model.locationofinterest.LocationOfInterest
 import com.google.android.ground.model.mutation.Mutation
 import com.google.android.ground.model.submission.Submission
-import com.google.android.ground.persistence.remote.RemoteDataEvent
 import com.google.android.ground.persistence.remote.RemoteDataStore
-import io.reactivex.Flowable
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -45,12 +43,6 @@ class FakeRemoteDataStore @Inject internal constructor() : RemoteDataStore {
 
   override suspend fun loadTermsOfService(): TermsOfService? = termsOfService?.getOrThrow()
 
-  // TODO(#1373): Delete once new LOI sync is implemented.
-  override fun loadLocationsOfInterestOnceAndStreamChanges(
-    survey: Survey
-  ): Flowable<RemoteDataEvent<LocationOfInterest>> =
-    Flowable.fromIterable(lois).map { RemoteDataEvent.loaded(it.id, it) }
-
   override suspend fun loadLocationsOfInterest(survey: Survey) = lois
 
   override suspend fun loadSubmissions(locationOfInterest: LocationOfInterest): List<Submission> {