use preference datastore to monitor syncJobStatus.

demo/src/main/java/com/google/android/fhir/demo/MainActivityViewModel.kt

@@ -24,13 +24,15 @@ import androidx.lifecycle.MutableLiveData
 import androidx.lifecycle.viewModelScope
 import androidx.work.Constraints
 import com.google.android.fhir.demo.data.DemoFhirSyncWorker
+import com.google.android.fhir.sync.CurrentSyncJobStatus
 import com.google.android.fhir.sync.PeriodicSyncConfiguration
+import com.google.android.fhir.sync.PeriodicSyncJobStatus
 import com.google.android.fhir.sync.RepeatInterval
 import com.google.android.fhir.sync.Sync
-import com.google.android.fhir.sync.SyncJobStatus
 import java.time.format.DateTimeFormatter
 import java.util.concurrent.TimeUnit
 import kotlinx.coroutines.InternalCoroutinesApi
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharingStarted
@@ -44,10 +46,14 @@ class MainActivityViewModel(application: Application) : AndroidViewModel(applica
   val lastSyncTimestampLiveData: LiveData<String>
     get() = _lastSyncTimestampLiveData
 
-  private val _pollState = MutableSharedFlow<SyncJobStatus>()
-  val pollState: Flow<SyncJobStatus>
+  private val _pollState = MutableSharedFlow<CurrentSyncJobStatus>()
+  val pollState: Flow<CurrentSyncJobStatus>
     get() = _pollState
 
+  private val _pollPeriodicSyncJobStatus = MutableSharedFlow<PeriodicSyncJobStatus>()
+  val pollPeriodicSyncJobStatus: Flow<PeriodicSyncJobStatus>
+    get() = _pollPeriodicSyncJobStatus
+
   init {
     viewModelScope.launch {
       Sync.periodicSync<DemoFhirSyncWorker>(
@@ -59,16 +65,23 @@ class MainActivityViewModel(application: Application) : AndroidViewModel(applica
             ),
         )
         .shareIn(this, SharingStarted.Eagerly, 10)
-        .collect { _pollState.emit(it) }
+        .collect { _pollPeriodicSyncJobStatus.emit(it) }
     }
   }
 
+  private var oneTimeSyncJob: Job? = null
+
   fun triggerOneTimeSync() {
-    viewModelScope.launch {
-      Sync.oneTimeSync<DemoFhirSyncWorker>(getApplication())
-        .shareIn(this, SharingStarted.Eagerly, 10)
-        .collect { _pollState.emit(it) }
-    }
+    // Cancels any ongoing sync job before starting a new one. Since this function may be called
+    // more than once, not canceling the ongoing job could result in the creation of multiple jobs
+    // that emit the same object.
+    oneTimeSyncJob?.cancel()
+    oneTimeSyncJob =
+      viewModelScope.launch {
+        Sync.oneTimeSync<DemoFhirSyncWorker>(getApplication())
+          .shareIn(this, SharingStarted.Eagerly, 0)
+          .collect { result -> result.let { _pollState.emit(it) } }
+      }
   }
 
   /** Emits last sync time. */

demo/src/main/java/com/google/android/fhir/demo/PatientListFragment.kt

@@ -44,6 +44,8 @@ import androidx.recyclerview.widget.RecyclerView
 import com.google.android.fhir.FhirEngine
 import com.google.android.fhir.demo.PatientListViewModel.PatientListViewModelFactory
 import com.google.android.fhir.demo.databinding.FragmentPatientListBinding
+import com.google.android.fhir.sync.CurrentSyncJobStatus
+import com.google.android.fhir.sync.LastSyncJobStatus
 import com.google.android.fhir.sync.SyncJobStatus
 import kotlin.math.roundToInt
 import kotlinx.coroutines.launch
@@ -105,6 +107,7 @@ class PatientListFragment : Fragment() {
 
     searchView = binding.search
     topBanner = binding.syncStatusContainer.linearLayoutSyncStatus
+    topBanner.visibility = View.GONE
     syncStatus = binding.syncStatusContainer.tvSyncingStatus
     syncPercent = binding.syncStatusContainer.tvSyncingPercent
     syncProgress = binding.syncStatusContainer.progressSyncing
@@ -155,28 +158,44 @@ class PatientListFragment : Fragment() {
       mainActivityViewModel.pollState.collect {
         Timber.d("onViewCreated: pollState Got status $it")
         when (it) {
-          is SyncJobStatus.Started -> {
-            Timber.i("Sync: ${it::class.java.simpleName}")
+          is CurrentSyncJobStatus.Running -> {
+            Timber.i("Sync: ${it::class.java.simpleName} with data ${it.inProgressSyncJob}")
             fadeInTopBanner(it)
           }
-          is SyncJobStatus.InProgress -> {
-            Timber.i("Sync: ${it::class.java.simpleName} with data $it")
-            fadeInTopBanner(it)
+          is CurrentSyncJobStatus.Succeeded -> {
+            Timber.i("Sync: ${it::class.java.simpleName} at ${it.succeededSyncJob.timestamp}")
+            patientListViewModel.searchPatientsByName(searchView.query.toString().trim())
+            mainActivityViewModel.updateLastSyncTimestamp()
+            fadeOutTopBanner(it)
           }
-          is SyncJobStatus.Finished -> {
-            Timber.i("Sync: ${it::class.java.simpleName} at ${it.timestamp}")
+          is CurrentSyncJobStatus.Failed -> {
+            Timber.i("Sync: ${it::class.java.simpleName} at ${it.failedSyncJob.timestamp}")
             patientListViewModel.searchPatientsByName(searchView.query.toString().trim())
             mainActivityViewModel.updateLastSyncTimestamp()
             fadeOutTopBanner(it)
           }
-          is SyncJobStatus.Failed -> {
-            Timber.i("Sync: ${it::class.java.simpleName} at ${it.timestamp}")
+          is CurrentSyncJobStatus.Enqueued -> {
+            Timber.i("Sync: Enqueued")
             patientListViewModel.searchPatientsByName(searchView.query.toString().trim())
             mainActivityViewModel.updateLastSyncTimestamp()
             fadeOutTopBanner(it)
           }
-          else -> {
-            Timber.i("Sync: Unknown state.")
+          CurrentSyncJobStatus.Cancelled -> TODO()
+        }
+      }
+    }
+
+    lifecycleScope.launch {
+      mainActivityViewModel.pollPeriodicSyncJobStatus.collect {
+        Timber.d("onViewCreated: pollState Got status $it")
+        if (it.currentSyncJobStatus is CurrentSyncJobStatus.Running) {
+          Timber.i(
+            "Sync: ${it.currentSyncJobStatus::class.java.simpleName} with data ${it.currentSyncJobStatus}",
+          )
+          fadeInTopBanner(it.currentSyncJobStatus)
+        } else {
+          it.lastSyncJobStatus?.let {
+            Timber.i("Sync: ${it::class.java.simpleName} at ${it.timestamp}")
             patientListViewModel.searchPatientsByName(searchView.query.toString().trim())
             mainActivityViewModel.updateLastSyncTimestamp()
             fadeOutTopBanner(it)
@@ -213,7 +232,7 @@ class PatientListFragment : Fragment() {
       .navigate(PatientListFragmentDirections.actionPatientListToAddPatientFragment())
   }
 
-  private fun fadeInTopBanner(state: SyncJobStatus) {
+  private fun fadeInTopBanner(state: CurrentSyncJobStatus) {
     if (topBanner.visibility != View.VISIBLE) {
       syncStatus.text = resources.getString(R.string.syncing).uppercase()
       syncPercent.text = ""
@@ -222,25 +241,35 @@ class PatientListFragment : Fragment() {
       topBanner.visibility = View.VISIBLE
       val animation = AnimationUtils.loadAnimation(topBanner.context, R.anim.fade_in)
       topBanner.startAnimation(animation)
-    } else if (state is SyncJobStatus.InProgress) {
+    } else if (
+      state is CurrentSyncJobStatus.Running && state.inProgressSyncJob is SyncJobStatus.InProgress
+    ) {
+      val inProgressState = state.inProgressSyncJob as? SyncJobStatus.InProgress
       val progress =
-        state
-          .let { it.completed.toDouble().div(it.total) }
-          .let { if (it.isNaN()) 0.0 else it }
-          .times(100)
-          .roundToInt()
-      "$progress% ${state.syncOperation.name.lowercase()}ed".also { syncPercent.text = it }
-      syncProgress.progress = progress
+        inProgressState
+          ?.let { it.completed.toDouble().div(it.total) }
+          ?.let { if (it.isNaN()) 0.0 else it }
+          ?.times(100)
+          ?.roundToInt()
+      "$progress% ${inProgressState?.syncOperation?.name?.lowercase()}ed"
+        .also { syncPercent.text = it }
+      syncProgress.progress = progress ?: 0
     }
   }
 
-  private fun fadeOutTopBanner(state: SyncJobStatus) {
-    if (state is SyncJobStatus.Finished) syncPercent.text = ""
-    syncProgress.visibility = View.GONE
+  private fun fadeOutTopBanner(state: CurrentSyncJobStatus) {
+    fadeOutTopBanner(state::class.java.simpleName.uppercase())
+  }
+
+  private fun fadeOutTopBanner(state: LastSyncJobStatus) {
+    fadeOutTopBanner(state::class.java.simpleName.uppercase())
+  }
 
+  private fun fadeOutTopBanner(statusText: String) {
+    syncPercent.text = ""
+    syncProgress.visibility = View.GONE
     if (topBanner.visibility == View.VISIBLE) {
-      "${resources.getString(R.string.sync).uppercase()} ${state::class.java.simpleName.uppercase()}"
-        .also { syncStatus.text = it }
+      "${resources.getString(R.string.sync).uppercase()} $statusText".also { syncStatus.text = it }
 
       val animation = AnimationUtils.loadAnimation(topBanner.context, R.anim.fade_out)
       topBanner.startAnimation(animation)

engine/benchmark/src/androidTest/java/com/google/android/fhir/benchmark/FhirSyncWorkerBenchmark.kt

@@ -22,6 +22,7 @@ import androidx.benchmark.junit4.measureRepeated
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import androidx.test.filters.SdkSuppress
+import androidx.work.Data
 import androidx.work.ListenableWorker
 import androidx.work.WorkerParameters
 import androidx.work.testing.TestListenableWorkerBuilder
@@ -138,7 +139,12 @@ class FhirSyncWorkerBenchmark {
   private fun oneTimeSync(numberPatients: Int, numberObservations: Int, numberEncounters: Int) =
     runBlocking {
       val context: Context = ApplicationProvider.getApplicationContext()
-      val worker = TestListenableWorkerBuilder<BenchmarkTestOneTimeSyncWorker>(context).build()
+      val inputData =
+        Data.Builder()
+          .putString("sync_status_preferences_datastore_key", "BenchmarkTestOneTimeSyncWorker")
+          .build()
+      val worker =
+        TestListenableWorkerBuilder<BenchmarkTestOneTimeSyncWorker>(context, inputData).build()
       setupMockServerDispatcher(numberPatients, numberObservations, numberEncounters)
       benchmarkRule.measureRepeated {
         runBlocking {

engine/src/androidTest/java/com/google/android/fhir/sync/SyncInstrumentedTest.kt

@@ -28,6 +28,7 @@ import com.google.android.fhir.FhirEngine
 import com.google.android.fhir.sync.upload.UploadStrategy
 import com.google.android.fhir.testing.TestDataSourceImpl
 import com.google.android.fhir.testing.TestDownloadManagerImpl
+import com.google.android.fhir.testing.TestFailingDatasource
 import com.google.android.fhir.testing.TestFhirEngineImpl
 import com.google.common.truth.Truth.assertThat
 import java.util.concurrent.TimeUnit
@@ -36,18 +37,21 @@ import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.flow.transformWhile
 import kotlinx.coroutines.runBlocking
-import org.junit.Ignore
 import org.junit.Test
 import org.junit.runner.RunWith
 
+/**
+ * Note : If you are running these tests on a local machine in Android Studio, make sure to clear
+ * the storage and cache of the `com.google.android.fhir.test` app on the emulator/device before
+ * running each test individually.
+ */
 @OptIn(ExperimentalCoroutinesApi::class)
 @RunWith(AndroidJUnit4::class)
-@Ignore("Flaky/fails due to https://github.com/google/android-fhir/issues/2046")
 class SyncInstrumentedTest {
 
   private val context: Context = InstrumentationRegistry.getInstrumentation().targetContext
 
-  class TestSyncWorker(appContext: Context, workerParams: WorkerParameters) :
+  open class TestSyncWorker(appContext: Context, workerParams: WorkerParameters) :
     FhirSyncWorker(appContext, workerParams) {
 
     override fun getFhirEngine(): FhirEngine = TestFhirEngineImpl
@@ -61,15 +65,20 @@ class SyncInstrumentedTest {
     override fun getUploadStrategy(): UploadStrategy = UploadStrategy.AllChangesSquashedBundlePut
   }
 
+  class TestSyncWorkerForDownloadFailing(appContext: Context, workerParams: WorkerParameters) :
+    TestSyncWorker(appContext, workerParams) {
+    override fun getDataSource(): DataSource = TestFailingDatasource
+  }
+
   @Test
   fun oneTime_worker_runs() {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
     val workManager = WorkManager.getInstance(context)
     runBlocking {
       Sync.oneTimeSync<TestSyncWorker>(context = context)
         .transformWhile {
-          emit(it is SyncJobStatus.Finished)
-          it !is SyncJobStatus.Finished
+          emit(it is CurrentSyncJobStatus.Succeeded)
+          it !is CurrentSyncJobStatus.Succeeded
         }
         .shareIn(this, SharingStarted.Eagerly, 5)
     }
@@ -78,6 +87,101 @@ class SyncInstrumentedTest {
       .isEqualTo(WorkInfo.State.SUCCEEDED)
   }
 
+  @Test
+  fun oneTime_worker_syncState() {
+    WorkManagerTestInitHelper.initializeTestWorkManager(context)
+    val states = mutableListOf<CurrentSyncJobStatus>()
+    runBlocking {
+      Sync.oneTimeSync<TestSyncWorker>(context = context)
+        .transformWhile {
+          states.add(it)
+          emit(it is CurrentSyncJobStatus.Succeeded)
+          it !is CurrentSyncJobStatus.Succeeded
+        }
+        .shareIn(this, SharingStarted.Eagerly, 5)
+    }
+    assertThat(states.first()).isInstanceOf(CurrentSyncJobStatus.Running::class.java)
+    assertThat(states.last()).isInstanceOf(CurrentSyncJobStatus.Succeeded::class.java)
+  }
+
+  @Test
+  fun oneTime_worker_failedSyncState() {
+    WorkManagerTestInitHelper.initializeTestWorkManager(context)
+    val states = mutableListOf<CurrentSyncJobStatus>()
+    runBlocking {
+      Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(context = context)
+        .transformWhile {
+          states.add(it)
+          emit(it is CurrentSyncJobStatus.Failed)
+          it !is CurrentSyncJobStatus.Failed
+        }
+        .shareIn(this, SharingStarted.Eagerly, 5)
+    }
+    assertThat(states.first()).isInstanceOf(CurrentSyncJobStatus.Running::class.java)
+    assertThat(states.last()).isInstanceOf(CurrentSyncJobStatus.Failed::class.java)
+  }
+
+  @Test
+  fun periodic_worker_periodicSyncState() {
+    WorkManagerTestInitHelper.initializeTestWorkManager(context)
+    val states = mutableListOf<PeriodicSyncJobStatus>()
+    // run and wait for periodic worker to finish
+    runBlocking {
+      Sync.periodicSync<TestSyncWorker>(
+          context = context,
+          periodicSyncConfiguration =
+            PeriodicSyncConfiguration(
+              syncConstraints = Constraints.Builder().build(),
+              repeat = RepeatInterval(interval = 15, timeUnit = TimeUnit.MINUTES),
+            ),
+        )
+        .transformWhile {
+          states.add(it)
+          emit(it)
+          it.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
+        }
+        .shareIn(this, SharingStarted.Eagerly, 5)
+    }
+
+    assertThat(states.first().currentSyncJobStatus)
+      .isInstanceOf(CurrentSyncJobStatus.Running::class.java)
+    assertThat(states.first().lastSyncJobStatus).isNull()
+    assertThat(states.last().currentSyncJobStatus)
+      .isInstanceOf(CurrentSyncJobStatus.Enqueued::class.java)
+    assertThat(states.last().lastSyncJobStatus)
+      .isInstanceOf(LastSyncJobStatus.Succeeded::class.java)
+  }
+
+  @Test
+  fun periodic_worker_failedPeriodicSyncState() {
+    WorkManagerTestInitHelper.initializeTestWorkManager(context)
+    val states = mutableListOf<PeriodicSyncJobStatus>()
+    // run and wait for periodic worker to finish
+    runBlocking {
+      Sync.periodicSync<TestSyncWorkerForDownloadFailing>(
+          context = context,
+          periodicSyncConfiguration =
+            PeriodicSyncConfiguration(
+              syncConstraints = Constraints.Builder().build(),
+              repeat = RepeatInterval(interval = 15, timeUnit = TimeUnit.MINUTES),
+            ),
+        )
+        .transformWhile {
+          states.add(it)
+          emit(it)
+          it.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
+        }
+        .shareIn(this, SharingStarted.Eagerly, 5)
+    }
+
+    assertThat(states.first().currentSyncJobStatus)
+      .isInstanceOf(CurrentSyncJobStatus.Running::class.java)
+    assertThat(states.first().lastSyncJobStatus).isNull()
+    assertThat(states.last().currentSyncJobStatus)
+      .isInstanceOf(CurrentSyncJobStatus.Enqueued::class.java)
+    assertThat(states.last().lastSyncJobStatus).isInstanceOf(LastSyncJobStatus.Failed::class.java)
+  }
+
   @Test
   fun periodic_worker_still_queued_to_run_after_oneTime_worker_started() {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
@@ -94,7 +198,7 @@ class SyncInstrumentedTest {
         )
         .transformWhile {
           emit(it)
-          it !is SyncJobStatus.Finished
+          it.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
         }
         .shareIn(this, SharingStarted.Eagerly, 5)
     }
@@ -110,7 +214,7 @@ class SyncInstrumentedTest {
       Sync.oneTimeSync<TestSyncWorker>(context = context)
         .transformWhile {
           emit(it)
-          it !is SyncJobStatus.Finished
+          it !is CurrentSyncJobStatus.Succeeded
         }
         .shareIn(this, SharingStarted.Eagerly, 5)
     }