[Rx -> Coroutine] Use coroutine when fetching LOIs from local db (#2022)

ground/src/main/java/com/google/android/ground/persistence/local/room/stores/RoomLocationOfInterestStore.kt

@@ -52,14 +52,6 @@ class RoomLocationOfInterestStore @Inject internal constructor() : LocalLocation
    * local database and returns a [Flowable] that continually emits the complete set anew any time
    * the underlying table changes (insertions, deletions, updates).
    */
-  override fun getLocationsOfInterestOnceAndStream(
-    survey: Survey
-  ): Flowable<Set<LocationOfInterest>> =
-    locationOfInterestDao
-      .findOnceAndStream(survey.id, EntityState.DEFAULT)
-      .map { toLocationsOfInterest(survey, it) }
-      .subscribeOn(schedulers.io())
-
   override fun findLocationsOfInterest(survey: Survey) =
     locationOfInterestDao.findByState(survey.id, EntityState.DEFAULT).map {
       toLocationsOfInterest(survey, it)

ground/src/main/java/com/google/android/ground/persistence/local/stores/LocalLocationOfInterestStore.kt

@@ -27,14 +27,6 @@ import kotlinx.coroutines.flow.Flow
 
 interface LocalLocationOfInterestStore :
   LocalMutationStore<LocationOfInterestMutation, LocationOfInterest> {
-  /**
-   * Returns a long-lived stream that emits the full set of LOIs for a survey on subscribe, and
-   * continues to return the full set each time a LOI is added/changed/removed.
-   */
-  fun getLocationsOfInterestOnceAndStream(
-    survey: Survey
-  ): @Cold(terminates = false) Flowable<Set<LocationOfInterest>>
-
   /**
    * Returns a main-safe flow that emits the full set of LOIs for a survey on subscribe, and
    * continues to return the full set each time a LOI is added/changed/removed.

ground/src/main/java/com/google/android/ground/repository/LocationOfInterestRepository.kt

@@ -37,7 +37,10 @@ import io.reactivex.Flowable
 import io.reactivex.Single
 import javax.inject.Inject
 import javax.inject.Singleton
-import kotlinx.coroutines.reactive.awaitFirst
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.map
 
 /**
  * Coordinates persistence and retrieval of [LocationOfInterest] instances from remote, local, and
@@ -122,23 +125,17 @@ constructor(
       MutationEntitySyncStatus.FAILED
     )
 
-  /** Returns a flowable of all [LocationOfInterest] for the given [Survey]. */
-  private fun getLocationsOfInterestOnceAndStream(
-    survey: Survey
-  ): Flowable<Set<LocationOfInterest>> = localLoiStore.getLocationsOfInterestOnceAndStream(survey)
-
-  fun getLocationsOfInterest(survey: Survey) = localLoiStore.findLocationsOfInterest(survey)
+  /** Returns a flow of all [LocationOfInterest] associated with the given [Survey]. */
+  fun getLocationsOfInterests(survey: Survey): Flow<Set<LocationOfInterest>> =
+    localLoiStore.findLocationsOfInterest(survey)
 
   /** Returns a list of geometries associated with the given [Survey]. */
   suspend fun getAllGeometries(survey: Survey): List<Geometry> =
-    getLocationsOfInterestOnceAndStream(survey).awaitFirst().map { it.geometry }
+    getLocationsOfInterests(survey).first().map { it.geometry }
 
-  /** Returns a flowable of all [LocationOfInterest] within the map bounds (viewport). */
-  fun getWithinBoundsOnceAndStream(
-    survey: Survey,
-    bounds: Bounds
-  ): Flowable<List<LocationOfInterest>> =
-    getLocationsOfInterestOnceAndStream(survey)
+  /** Returns a flow of all [LocationOfInterest] within the map bounds (viewport). */
+  fun getWithinBounds(survey: Survey, bounds: Bounds): Flow<List<LocationOfInterest>> =
+    getLocationsOfInterests(survey)
       .map { lois -> lois.filter { bounds.contains(it.geometry) } }
       .distinctUntilChanged()
 }

ground/src/main/java/com/google/android/ground/ui/home/mapcontainer/HomeScreenMapContainerViewModel.kt

@@ -56,7 +56,6 @@ import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.flow.stateIn
-import kotlinx.coroutines.reactive.asFlow
 import timber.log.Timber
 
 @OptIn(ExperimentalCoroutinesApi::class)
@@ -128,7 +127,7 @@ internal constructor(
         .flatMapLatest { (survey, isZoomedIn) ->
           val bounds = currentCameraPosition.value?.bounds
           if (bounds == null || survey == null || !isZoomedIn) flowOf(listOf())
-          else loiRepository.getWithinBoundsOnceAndStream(survey, bounds).asFlow()
+          else loiRepository.getWithinBounds(survey, bounds)
         }
         .stateIn(viewModelScope, SharingStarted.Lazily, listOf())
 
@@ -179,7 +178,7 @@ internal constructor(
   fun getZoomThresholdCrossed(): Observable<Nil> = zoomThresholdCrossed
 
   private fun getLocationOfInterestFeatures(survey: Survey): Flow<Set<Feature>> =
-    loiRepository.getLocationsOfInterest(survey).map {
+    loiRepository.getLocationsOfInterests(survey).map {
       it.map { loi -> loi.toFeature() }.toPersistentSet()
     }
 

ground/src/test/java/com/google/android/ground/persistence/local/LocalDataStoreTests.kt

@@ -173,14 +173,16 @@ class LocalDataStoreTests : BaseHiltTest() {
   }
 
   @Test
-  fun testGetLoisOnceAndStream() = runWithTestDispatcher {
+  fun testFindLocationsOfInterest() = runWithTestDispatcher {
     localUserStore.insertOrUpdateUser(TEST_USER)
     localSurveyStore.insertOrUpdateSurvey(TEST_SURVEY)
-    val subscriber = localLoiStore.getLocationsOfInterestOnceAndStream(TEST_SURVEY).test()
-    subscriber.assertValue(setOf())
     localLoiStore.applyAndEnqueue(TEST_LOI_MUTATION)
+
     val loi = localLoiStore.getLocationOfInterest(TEST_SURVEY, "loi id").blockingGet()
-    subscriber.assertValueSet(setOf(setOf(), setOf(loi)))
+
+    localLoiStore.findLocationsOfInterest(TEST_SURVEY).test {
+      assertThat(expectMostRecentItem()).isEqualTo(setOf(loi))
+    }
   }
 
   @Test
@@ -312,11 +314,12 @@ class LocalDataStoreTests : BaseHiltTest() {
     localSurveyStore.insertOrUpdateSurvey(TEST_SURVEY)
     localLoiStore.applyAndEnqueue(TEST_LOI_MUTATION)
     localSubmissionStore.applyAndEnqueue(TEST_SUBMISSION_MUTATION)
-    val subscriber = localLoiStore.getLocationsOfInterestOnceAndStream(TEST_SURVEY).test()
 
     // Assert that one LOI is streamed.
     val loi = localLoiStore.getLocationOfInterest(TEST_SURVEY, "loi id").blockingGet()
-    subscriber.assertValueAt(0, setOf(loi))
+    localLoiStore.findLocationsOfInterest(TEST_SURVEY).test {
+      assertThat(expectMostRecentItem()).isEqualTo(setOf(loi))
+    }
     val mutation = TEST_LOI_MUTATION.copy(id = null, type = Mutation.Type.DELETE)
 
     // Calling applyAndEnqueue marks the local LOI as deleted.
@@ -329,7 +332,9 @@ class LocalDataStoreTests : BaseHiltTest() {
     }
 
     // Verify that the local LOI is now removed from the latest LOI stream.
-    subscriber.assertValueAt(1, setOf())
+    localLoiStore.findLocationsOfInterest(TEST_SURVEY).test {
+      assertThat(expectMostRecentItem()).isEmpty()
+    }
 
     // After successful remote sync, delete LOI is called by LocalMutationSyncWorker.
     localLoiStore.deleteLocationOfInterest("loi id")

ground/src/test/java/com/google/android/ground/repository/LocationOfInterestRepositoryTest.kt

@@ -15,27 +15,34 @@
  */
 package com.google.android.ground.repository
 
+import app.cash.turbine.test
 import com.google.android.ground.BaseHiltTest
 import com.google.android.ground.domain.usecases.survey.ActivateSurveyUseCase
-import com.google.android.ground.model.geometry.*
+import com.google.android.ground.model.geometry.Coordinates
+import com.google.android.ground.model.geometry.LinearRing
+import com.google.android.ground.model.geometry.Point
+import com.google.android.ground.model.geometry.Polygon
 import com.google.android.ground.model.mutation.Mutation.Type.CREATE
 import com.google.android.ground.persistence.sync.MutationSyncWorkManager
 import com.google.android.ground.ui.map.Bounds
+import com.google.common.truth.Truth.assertThat
 import com.sharedtest.FakeData
 import com.sharedtest.persistence.remote.FakeRemoteDataStore
 import com.sharedtest.system.auth.FakeAuthenticationManager
 import dagger.hilt.android.testing.BindValue
 import dagger.hilt.android.testing.HiltAndroidTest
-import java.util.*
 import javax.inject.Inject
 import kotlin.test.assertFailsWith
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.advanceUntilIdle
 import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
-import org.mockito.*
-import org.mockito.Mockito.*
+import org.mockito.ArgumentMatchers.anyString
+import org.mockito.Mock
+import org.mockito.Mockito.`when`
+import org.mockito.kotlin.times
+import org.mockito.kotlin.verify
 import org.robolectric.RobolectricTestRunner
 
 @HiltAndroidTest
@@ -123,44 +130,43 @@ class LocationOfInterestRepositoryTest : BaseHiltTest() {
   // TODO(#1373): Add tests for getLocationsOfInterest once new LOI sync implemented.
 
   @Test
-  fun testLoiWithinBounds_whenOutOfBounds_returnsEmptyList() {
+  fun testLoiWithinBounds_whenOutOfBounds_returnsEmptyList() = runWithTestDispatcher {
     val southwest = Coordinates(-60.0, -60.0)
     val northeast = Coordinates(-50.0, -50.0)
 
-    locationOfInterestRepository
-      .getWithinBoundsOnceAndStream(TEST_SURVEY, Bounds(southwest, northeast))
-      .test()
-      .assertValues(listOf())
+    locationOfInterestRepository.getWithinBounds(TEST_SURVEY, Bounds(southwest, northeast)).test {
+      assertThat(expectMostRecentItem()).isEmpty()
+    }
   }
 
   @Test
-  fun testLoiWithinBounds_whenSomeLOIsInsideBounds_returnsPartialList() {
+  fun testLoiWithinBounds_whenSomeLOIsInsideBounds_returnsPartialList() = runWithTestDispatcher {
     val southwest = Coordinates(-20.0, -20.0)
     val northeast = Coordinates(-10.0, -10.0)
 
-    locationOfInterestRepository
-      .getWithinBoundsOnceAndStream(TEST_SURVEY, Bounds(southwest, northeast))
-      .test()
-      .assertValues(listOf(TEST_POINT_OF_INTEREST_1, TEST_AREA_OF_INTEREST_1))
+    locationOfInterestRepository.getWithinBounds(TEST_SURVEY, Bounds(southwest, northeast)).test {
+      assertThat(expectMostRecentItem())
+        .isEqualTo(listOf(TEST_POINT_OF_INTEREST_1, TEST_AREA_OF_INTEREST_1))
+    }
   }
 
   @Test
-  fun testLoiWithinBounds_whenAllLOIsInsideBounds_returnsCompleteList() {
+  fun testLoiWithinBounds_whenAllLOIsInsideBounds_returnsCompleteList() = runWithTestDispatcher {
     val southwest = Coordinates(-20.0, -20.0)
     val northeast = Coordinates(20.0, 20.0)
 
-    locationOfInterestRepository
-      .getWithinBoundsOnceAndStream(TEST_SURVEY, Bounds(southwest, northeast))
-      .test()
-      .assertValues(
-        listOf(
-          TEST_POINT_OF_INTEREST_1,
-          TEST_POINT_OF_INTEREST_2,
-          TEST_POINT_OF_INTEREST_3,
-          TEST_AREA_OF_INTEREST_1,
-          TEST_AREA_OF_INTEREST_2
+    locationOfInterestRepository.getWithinBounds(TEST_SURVEY, Bounds(southwest, northeast)).test {
+      assertThat(expectMostRecentItem())
+        .isEqualTo(
+          listOf(
+            TEST_POINT_OF_INTEREST_1,
+            TEST_POINT_OF_INTEREST_2,
+            TEST_POINT_OF_INTEREST_3,
+            TEST_AREA_OF_INTEREST_1,
+            TEST_AREA_OF_INTEREST_2
+          )
         )
-      )
+    }
   }
 
   companion object {