Revert "Use coroutine in Navigator"

ground/src/main/java/com/google/android/ground/MainActivity.kt

@@ -20,19 +20,18 @@ import android.app.ProgressDialog
 import android.content.Intent
 import android.os.Bundle
 import androidx.core.view.WindowInsetsCompat
-import androidx.lifecycle.lifecycleScope
 import androidx.navigation.NavDirections
 import androidx.navigation.fragment.NavHostFragment
 import com.google.android.ground.databinding.MainActBinding
 import com.google.android.ground.repository.UserRepository
 import com.google.android.ground.rx.RxAutoDispose.autoDisposable
+import com.google.android.ground.rx.Schedulers
 import com.google.android.ground.system.ActivityStreams
 import com.google.android.ground.system.SettingsManager
 import com.google.android.ground.ui.common.*
 import dagger.hilt.android.AndroidEntryPoint
 import java8.util.function.Consumer
 import javax.inject.Inject
-import kotlinx.coroutines.launch
 import timber.log.Timber
 
 /**
@@ -52,6 +51,8 @@ class MainActivity : Hilt_MainActivity() {
 
   @Inject lateinit var popups: EphemeralPopups
 
+  @Inject lateinit var schedulers: Schedulers
+
   private lateinit var viewModel: MainViewModel
   private lateinit var navHostFragment: NavHostFragment
 
@@ -68,6 +69,22 @@ class MainActivity : Hilt_MainActivity() {
       callback.accept(this)
     }
 
+    navigator
+      .getNavigateRequests()
+      .observeOn(schedulers.ui())
+      .`as`(autoDisposable(this))
+      .subscribe { navDirections: NavDirections -> onNavigate(navDirections) }
+
+    navigator
+      .getNavigateUpRequests()
+      .observeOn(schedulers.ui())
+      .`as`(autoDisposable(this))
+      .subscribe { navigateUp() }
+
+    navigator.getFinishRequests().observeOn(schedulers.ui()).`as`(autoDisposable(this)).subscribe {
+      finish()
+    }
+
     val binding = MainActBinding.inflate(layoutInflater)
     setContentView(binding.root)
 
@@ -78,10 +95,6 @@ class MainActivity : Hilt_MainActivity() {
     viewModel.signInProgressDialogVisibility.observe(this) { visible: Boolean ->
       onSignInProgress(visible)
     }
-
-    lifecycleScope.launch { navigator.getNavigateRequests().collect { onNavigate(it) } }
-    lifecycleScope.launch { navigator.getNavigateUpRequests().collect { navigateUp() } }
-    lifecycleScope.launch { navigator.getFinishRequests().collect { finish() } }
   }
 
   override fun onWindowInsetChanged(insets: WindowInsetsCompat) {

ground/src/main/java/com/google/android/ground/ui/common/Navigator.kt

@@ -16,42 +16,44 @@
 package com.google.android.ground.ui.common
 
 import androidx.navigation.NavDirections
+import com.google.android.ground.rx.annotations.Hot
+import io.reactivex.Observable
+import io.reactivex.subjects.PublishSubject
+import io.reactivex.subjects.Subject
 import javax.inject.Inject
 import javax.inject.Singleton
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers.Main
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.launch
 
 /**
  * Responsible for abstracting navigation from fragment to fragment. Exposes various actions to
- * ViewModels that cause a NavDirections to be emitted to the flow.
+ * ViewModels that cause a NavDirections to be emitted to the observer (in this case, the [ ], which
+ * is expected to pass it to the current [ ].
  */
 @Singleton
 class Navigator @Inject constructor() {
-  private val navigateRequests: MutableStateFlow<NavDirections?> = MutableStateFlow(null)
-  private val navigateUpRequests: MutableStateFlow<Any?> = MutableStateFlow(null)
-  private val finishRequests: MutableStateFlow<Any?> = MutableStateFlow(null)
+  private val navigateRequests: @Hot Subject<NavDirections> = PublishSubject.create()
+  private val navigateUpRequests: @Hot Subject<Any> = PublishSubject.create()
+  private val finishRequests: @Hot Subject<Any> = PublishSubject.create()
 
   /** Stream of navigation requests for fulfillment by the view layer. */
-  fun getNavigateRequests(): Flow<NavDirections> = navigateRequests.filterNotNull()
+  fun getNavigateRequests(): Observable<NavDirections> = navigateRequests
 
-  fun getNavigateUpRequests(): Flow<Any> = navigateUpRequests.filterNotNull()
+  fun getNavigateUpRequests(): Observable<Any> = navigateUpRequests
 
-  fun getFinishRequests(): Flow<Any> = finishRequests.filterNotNull()
+  fun getFinishRequests(): Observable<Any> = finishRequests
 
   /** Navigates up one level on the back stack. */
   fun navigateUp() {
-    CoroutineScope(Main).launch { navigateUpRequests.emit(Any()) }
+    navigateUpRequests.onNext(Any())
   }
 
   fun navigate(directions: NavDirections) {
-    CoroutineScope(Main).launch { navigateRequests.emit(directions) }
+    CoroutineScope(Main).launch { navigateRequests.onNext(directions) }
   }
 
   fun finishApp() {
-    CoroutineScope(Main).launch { finishRequests.emit(Any()) }
+    finishRequests.onNext(Any())
   }
 }

ground/src/test/java/com/google/android/ground/MainViewModelTest.kt

@@ -18,7 +18,6 @@ package com.google.android.ground
 import android.content.SharedPreferences
 import android.os.Looper
 import androidx.navigation.NavDirections
-import app.cash.turbine.test
 import com.google.android.ground.persistence.local.room.LocalDataStoreException
 import com.google.android.ground.repository.TermsOfServiceRepository
 import com.google.android.ground.repository.UserRepository
@@ -33,6 +32,7 @@ import com.sharedtest.TestObservers.observeUntilFirstChange
 import com.sharedtest.persistence.remote.FakeRemoteDataStore
 import com.sharedtest.system.auth.FakeAuthenticationManager
 import dagger.hilt.android.testing.HiltAndroidTest
+import io.reactivex.observers.TestObserver
 import javax.inject.Inject
 import kotlin.test.assertFailsWith
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -56,11 +56,16 @@ class MainViewModelTest : BaseHiltTest() {
   @Inject lateinit var tosRepository: TermsOfServiceRepository
   @Inject lateinit var userRepository: UserRepository
 
+  private lateinit var navDirectionsTestObserver: TestObserver<NavDirections>
+
   @Before
   override fun setUp() {
     super.setUp()
 
     fakeAuthenticationManager.setUser(FakeData.USER)
+
+    // Subscribe to navigation requests
+    navDirectionsTestObserver = navigator.getNavigateRequests().test()
   }
 
   private fun setupUserPreferences() {
@@ -84,16 +89,11 @@ class MainViewModelTest : BaseHiltTest() {
     assertThat(viewModel.signInProgressDialogVisibility.value).isEqualTo(visible)
   }
 
-  private fun verifyNavigationRequested(navDirections: NavDirections? = null) =
-    runWithTestDispatcher {
-      navigator.getNavigateRequests().test {
-        if (navDirections == null) {
-          expectNoEvents()
-        } else {
-          assertThat(expectMostRecentItem()).isEqualTo(navDirections)
-        }
-      }
-    }
+  private fun verifyNavigationRequested(vararg navDirections: NavDirections) {
+    navDirectionsTestObserver.assertNoErrors()
+    navDirectionsTestObserver.assertNotComplete()
+    navDirectionsTestObserver.assertValues(*navDirections)
+  }
 
   @Test
   fun testSignInStateChanged_onSignedOut() {

ground/src/test/java/com/google/android/ground/ui/home/HomeScreenFragmentTest.kt

@@ -27,22 +27,21 @@ import androidx.test.espresso.matcher.ViewMatchers
 import androidx.test.espresso.matcher.ViewMatchers.isEnabled
 import androidx.test.espresso.matcher.ViewMatchers.withId
 import androidx.test.espresso.matcher.ViewMatchers.withText
-import app.cash.turbine.test
 import com.google.android.ground.BaseHiltTest
 import com.google.android.ground.R
 import com.google.android.ground.launchFragmentInHiltContainer
 import com.google.android.ground.model.Survey
 import com.google.android.ground.model.imagery.TileSource
 import com.google.android.ground.repository.SurveyRepository
 import com.google.android.ground.ui.common.Navigator
-import com.google.common.truth.Truth.assertThat
 import com.sharedtest.FakeData
 import com.squareup.picasso.Picasso
 import dagger.hilt.android.testing.HiltAndroidTest
 import javax.inject.Inject
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.advanceUntilIdle
 import org.hamcrest.CoreMatchers.not
+import org.junit.Assert.*
 import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -154,7 +153,6 @@ class HomeScreenFragmentTest : AbstractHomeScreenFragmentTest() {
   }
 }
 
-@OptIn(ExperimentalCoroutinesApi::class)
 @HiltAndroidTest
 @RunWith(ParameterizedRobolectricTestRunner::class)
 class NavigationDrawerItemClickTest(
@@ -165,13 +163,13 @@ class NavigationDrawerItemClickTest(
   @Inject lateinit var navigator: Navigator
 
   @Test
-  fun clickDrawerMenuItem() = runWithTestDispatcher {
+  fun clickDrawerMenuItem() {
+    val navDirectionsTestObserver = navigator.getNavigateRequests().test()
+
     openDrawer()
     onView(withText(menuItemLabel)).check(matches(isEnabled())).perform(click())
 
-    navigator.getNavigateRequests().test {
-      assertThat(expectMostRecentItem()).isEqualTo(expectedNavDirection)
-    }
+    navDirectionsTestObserver.assertValue(expectedNavDirection)
     verifyDrawerClosed()
   }
 

ground/src/test/java/com/google/android/ground/ui/home/mapcontainer/HomeScreenMapContainerFragmentTest.kt

@@ -20,21 +20,18 @@ import androidx.test.espresso.action.ViewActions.click
 import androidx.test.espresso.assertion.ViewAssertions.matches
 import androidx.test.espresso.matcher.ViewMatchers.isEnabled
 import androidx.test.espresso.matcher.ViewMatchers.withId
-import app.cash.turbine.test
 import com.google.android.ground.*
 import com.google.android.ground.NavGraphDirections.ShowMapTypeDialogFragment
 import com.google.android.ground.ui.common.Navigator
 import com.google.android.ground.ui.map.MapType
-import com.google.common.truth.Truth.assertThat
 import dagger.hilt.android.testing.HiltAndroidTest
 import javax.inject.Inject
-import kotlinx.coroutines.ExperimentalCoroutinesApi
+import org.junit.Assert.*
 import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.robolectric.RobolectricTestRunner
 
-@OptIn(ExperimentalCoroutinesApi::class)
 @HiltAndroidTest
 @RunWith(RobolectricTestRunner::class)
 class HomeScreenMapContainerFragmentTest : BaseHiltTest() {
@@ -48,13 +45,13 @@ class HomeScreenMapContainerFragmentTest : BaseHiltTest() {
   }
 
   @Test
-  fun clickMapType_launchesMapTypeDialogFragment() = runWithTestDispatcher {
+  fun clickMapType_launchesMapTypeDialogFragment() {
+    val navDirectionsTestObserver = navigator.getNavigateRequests().test()
+
     onView(withId(R.id.map_type_btn)).perform(click()).check(matches(isEnabled()))
 
-    navigator.getNavigateRequests().test {
-      val result = expectMostRecentItem()
-      assertThat(result).isInstanceOf(ShowMapTypeDialogFragment::class.java)
-      assertThat((result as ShowMapTypeDialogFragment).mapTypes).isEqualTo(MapType.values())
+    navDirectionsTestObserver.assertValue {
+      it is ShowMapTypeDialogFragment && it.mapTypes.contentEquals(MapType.values())
     }
   }
 }

ground/src/test/java/com/google/android/ground/ui/tos/TermsOfServiceFragmentTest.kt

@@ -20,7 +20,6 @@ import androidx.test.espresso.Espresso.onView
 import androidx.test.espresso.action.ViewActions.click
 import androidx.test.espresso.assertion.ViewAssertions.matches
 import androidx.test.espresso.matcher.ViewMatchers.*
-import app.cash.turbine.test
 import com.google.android.ground.BaseHiltTest
 import com.google.android.ground.R
 import com.google.android.ground.launchFragmentInHiltContainer
@@ -30,13 +29,11 @@ import com.google.android.ground.ui.surveyselector.SurveySelectorFragmentDirecti
 import com.google.common.truth.Truth.assertThat
 import dagger.hilt.android.testing.HiltAndroidTest
 import javax.inject.Inject
-import kotlinx.coroutines.ExperimentalCoroutinesApi
 import org.hamcrest.Matchers.not
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.robolectric.RobolectricTestRunner
 
-@OptIn(ExperimentalCoroutinesApi::class)
 @HiltAndroidTest
 @RunWith(RobolectricTestRunner::class)
 class TermsOfServiceFragmentTest : BaseHiltTest() {
@@ -72,19 +69,19 @@ class TermsOfServiceFragmentTest : BaseHiltTest() {
   }
 
   @Test
-  fun agreeButton_whenPressed_shouldUpdatePrefAndNavigate() = runWithTestDispatcher {
+  fun agreeButton_whenPressed_shouldUpdatePrefAndNavigate() {
     launchFragmentInHiltContainer<TermsOfServiceFragment>(bundleOf())
 
     assertThat(termsOfServiceRepository.isTermsOfServiceAccepted).isFalse()
+    val navDirectionsTestObserver = navigator.getNavigateRequests().test()
 
     onView(withId(R.id.agreeCheckBox)).perform(click())
     onView(withId(R.id.agreeButton)).perform(click())
 
     assertThat(termsOfServiceRepository.isTermsOfServiceAccepted).isTrue()
-    navigator.getNavigateRequests().test {
-      assertThat(expectMostRecentItem())
-        .isEqualTo(SurveySelectorFragmentDirections.showSurveySelectorScreen(true))
-    }
+    navDirectionsTestObserver.assertValue(
+      SurveySelectorFragmentDirections.showSurveySelectorScreen(true)
+    )
   }
 
   @Test