Configurable Bundle size to upload to the server (#1584)

* Users can provide an upload configuration to limit the number of resources in each bundle to upload to the server

* Review changes: Added upload config as param for SyncJob

* Review comments: Refactored code to just have Size based splitter

Co-authored-by: PallaviGanorkar <[email protected]>

Author: aditya-07

engine/src/main/java/com/google/android/fhir/sync/Config.kt

@@ -34,6 +34,9 @@ typealias ParamMap = Map<String, String>
 /** Constant for the Greater Than Search Prefix */
 @PublishedApi internal const val GREATER_THAN_PREFIX = "gt"
 
+/** Constant for the default number of resource entries in a singe Bundle for upload. */
+const val DEFAULT_BUNDLE_SIZE = 500
+
 val defaultRetryConfiguration =
   RetryConfiguration(BackoffCriteria(BackoffPolicy.LINEAR, 30, TimeUnit.SECONDS), 3)
 
@@ -104,3 +107,15 @@ data class BackoffCriteria(
   /** The time unit for [backoffDelay] */
   val timeUnit: TimeUnit
 )
+
+/**
+ * Configuration for max number of resources to be uploaded in a Bundle.The default size is
+ * [DEFAULT_BUNDLE_SIZE].
+ */
+data class UploadConfiguration(
+  /**
+   * Number of [Resource]s to be added in a singe [Bundle] for upload and default is
+   * [DEFAULT_BUNDLE_SIZE]
+   */
+  val uploadBundleSize: Int = DEFAULT_BUNDLE_SIZE
+)

engine/src/main/java/com/google/android/fhir/sync/FhirSyncWorker.kt

@@ -26,6 +26,10 @@ import com.google.android.fhir.FhirEngineProvider
 import com.google.android.fhir.OffsetDateTimeTypeAdapter
 import com.google.android.fhir.sync.Result.Error
 import com.google.android.fhir.sync.Result.Success
+import com.google.android.fhir.sync.download.DownloaderImpl
+import com.google.android.fhir.sync.upload.BundleUploader
+import com.google.android.fhir.sync.upload.LocalChangesPaginator
+import com.google.android.fhir.sync.upload.TransactionBundleGenerator
 import com.google.gson.ExclusionStrategy
 import com.google.gson.FieldAttributes
 import com.google.gson.GsonBuilder
@@ -45,6 +49,12 @@ abstract class FhirSyncWorker(appContext: Context, workerParams: WorkerParameter
   abstract fun getDownloadWorkManager(): DownloadWorkManager
   abstract fun getConflictResolver(): ConflictResolver
 
+  /**
+   * Configuration defining the max upload Bundle size (in terms to number of resources in a Bundle)
+   * and optionally defining the order of Resources.
+   */
+  open fun getUploadConfiguration(): UploadConfiguration = UploadConfiguration()
+
   private val gson =
     GsonBuilder()
       .registerTypeAdapter(OffsetDateTime::class.java, OffsetDateTimeTypeAdapter().nullSafe())
@@ -65,14 +75,6 @@ abstract class FhirSyncWorker(appContext: Context, workerParams: WorkerParameter
           )
         )
 
-    val fhirSynchronizer =
-      FhirSynchronizer(
-        applicationContext,
-        getFhirEngine(),
-        dataSource,
-        getDownloadWorkManager(),
-        conflictResolver = getConflictResolver()
-      )
     val flow = MutableSharedFlow<State>()
 
     val job =
@@ -87,11 +89,21 @@ abstract class FhirSyncWorker(appContext: Context, workerParams: WorkerParameter
         }
       }
 
-    fhirSynchronizer.subscribe(flow)
-
     Timber.v("Subscribed to flow for progress")
-
-    val result = fhirSynchronizer.synchronize()
+    val result =
+      FhirSynchronizer(
+          applicationContext,
+          getFhirEngine(),
+          BundleUploader(
+            dataSource,
+            TransactionBundleGenerator.getDefault(),
+            LocalChangesPaginator.create(getUploadConfiguration())
+          ),
+          DownloaderImpl(dataSource, getDownloadWorkManager()),
+          getConflictResolver()
+        )
+        .apply { subscribe(flow) }
+        .synchronize()
     val output = buildOutput(result)
 
     // await/join is needed to collect states completely

engine/src/main/java/com/google/android/fhir/sync/FhirSynchronizer.kt

@@ -19,9 +19,6 @@ package com.google.android.fhir.sync
 import android.content.Context
 import com.google.android.fhir.DatastoreUtil
 import com.google.android.fhir.FhirEngine
-import com.google.android.fhir.sync.download.DownloaderImpl
-import com.google.android.fhir.sync.upload.BundleUploader
-import com.google.android.fhir.sync.upload.TransactionBundleGenerator
 import java.time.OffsetDateTime
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.collect
@@ -51,11 +48,8 @@ data class ResourceSyncException(val resourceType: ResourceType, val exception:
 internal class FhirSynchronizer(
   context: Context,
   private val fhirEngine: FhirEngine,
-  private val dataSource: DataSource,
-  private val downloadManager: DownloadWorkManager,
-  private val uploader: Uploader =
-    BundleUploader(dataSource, TransactionBundleGenerator.getDefault()),
-  private val downloader: Downloader = DownloaderImpl(dataSource, downloadManager),
+  private val uploader: Uploader,
+  private val downloader: Downloader,
   private val conflictResolver: ConflictResolver
 ) {
   private var syncState: MutableSharedFlow<State>? = null

engine/src/main/java/com/google/android/fhir/sync/Sync.kt

@@ -25,6 +25,10 @@ import androidx.work.PeriodicWorkRequest
 import androidx.work.WorkManager
 import com.google.android.fhir.FhirEngine
 import com.google.android.fhir.FhirEngineProvider
+import com.google.android.fhir.sync.download.DownloaderImpl
+import com.google.android.fhir.sync.upload.BundleUploader
+import com.google.android.fhir.sync.upload.LocalChangesPaginator
+import com.google.android.fhir.sync.upload.TransactionBundleGenerator
 import org.hl7.fhir.r4.model.ResourceType
 
 object Sync {
@@ -42,10 +46,21 @@ object Sync {
     context: Context,
     fhirEngine: FhirEngine,
     downloadManager: DownloadWorkManager,
+    uploadConfiguration: UploadConfiguration = UploadConfiguration(),
     resolver: ConflictResolver
   ): Result {
     return FhirEngineProvider.getDataSource(context)?.let {
-      FhirSynchronizer(context, fhirEngine, it, downloadManager, conflictResolver = resolver)
+      FhirSynchronizer(
+          context,
+          fhirEngine,
+          BundleUploader(
+            it,
+            TransactionBundleGenerator.getDefault(),
+            LocalChangesPaginator.create(uploadConfiguration)
+          ),
+          DownloaderImpl(it, downloadManager),
+          resolver
+        )
         .synchronize()
     }
       ?: Result.Error(

engine/src/main/java/com/google/android/fhir/sync/SyncJob.kt

@@ -34,7 +34,8 @@ interface SyncJob {
     fhirEngine: FhirEngine,
     downloadManager: DownloadWorkManager,
     resolver: ConflictResolver,
-    subscribeTo: MutableSharedFlow<State>?
+    subscribeTo: MutableSharedFlow<State>?,
+    uploadConfiguration: UploadConfiguration = UploadConfiguration()
   ): Result
 
   fun workInfoFlow(): Flow<WorkInfo>

engine/src/main/java/com/google/android/fhir/sync/SyncJobImpl.kt

@@ -26,6 +26,10 @@ import com.google.android.fhir.DatastoreUtil
 import com.google.android.fhir.FhirEngine
 import com.google.android.fhir.FhirEngineProvider
 import com.google.android.fhir.OffsetDateTimeTypeAdapter
+import com.google.android.fhir.sync.download.DownloaderImpl
+import com.google.android.fhir.sync.upload.BundleUploader
+import com.google.android.fhir.sync.upload.LocalChangesPaginator
+import com.google.android.fhir.sync.upload.TransactionBundleGenerator
 import com.google.gson.GsonBuilder
 import java.time.OffsetDateTime
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -101,10 +105,21 @@ class SyncJobImpl(private val context: Context) : SyncJob {
     fhirEngine: FhirEngine,
     downloadManager: DownloadWorkManager,
     resolver: ConflictResolver,
-    subscribeTo: MutableSharedFlow<State>?
+    subscribeTo: MutableSharedFlow<State>?,
+    uploadConfiguration: UploadConfiguration
   ): Result {
     return FhirEngineProvider.getDataSource(context)?.let {
-      FhirSynchronizer(context, fhirEngine, it, downloadManager, conflictResolver = resolver)
+      FhirSynchronizer(
+          context,
+          fhirEngine,
+          BundleUploader(
+            it,
+            TransactionBundleGenerator.getDefault(),
+            LocalChangesPaginator.create(uploadConfiguration)
+          ),
+          DownloaderImpl(it, downloadManager),
+          resolver
+        )
         .apply { if (subscribeTo != null) subscribe(subscribeTo) }
         .synchronize()
     }

engine/src/main/java/com/google/android/fhir/sync/upload/BundleUploader.kt

@@ -33,13 +33,15 @@ import org.hl7.fhir.r4.model.ResourceType
 /** [Uploader] implementation to work with Fhir [Bundle]. */
 internal class BundleUploader(
   private val dataSource: DataSource,
-  private val bundleGenerator: TransactionBundleGenerator
+  private val bundleGenerator: TransactionBundleGenerator,
+  private val localChangesPaginator: LocalChangesPaginator
 ) : Uploader {
 
   override suspend fun upload(
     localChanges: List<LocalChange>,
   ): Flow<UploadResult> = flow {
-    bundleGenerator.generate(listOf(localChanges)).forEach { (bundle, localChangeTokens) ->
+    bundleGenerator.generate(localChangesPaginator.page(localChanges)).forEach {
+      (bundle, localChangeTokens) ->
       try {
         val response = dataSource.upload(bundle)
         emit(getUploadResult(response, localChangeTokens))

engine/src/main/java/com/google/android/fhir/sync/upload/LocalChangesPaginator.kt

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.android.fhir.sync.upload
+
+import com.google.android.fhir.LocalChange
+import com.google.android.fhir.sync.UploadConfiguration
+
+/** Splits the [List]<[LocalChange]> into smaller chunks. */
+internal fun interface LocalChangesPaginator {
+  fun page(list: List<LocalChange>): List<List<LocalChange>>
+
+  companion object Factory {
+
+    fun create(uploadConfiguration: UploadConfiguration): LocalChangesPaginator {
+      require(uploadConfiguration.uploadBundleSize > 0) {
+        "UploadConfiguration.uploadBundleSize ${uploadConfiguration.uploadBundleSize} must be greater than zero."
+      }
+
+      return SizeBasedLocalChangesPaginator(uploadConfiguration.uploadBundleSize)
+    }
+
+    val DEFAULT = create(UploadConfiguration())
+  }
+}
+
+/**
+ * Splits the [List]<[LocalChange]> into equal chunks of provided [size]. The last chunk may have
+ * fewer elements than the [size].
+ */
+internal class SizeBasedLocalChangesPaginator(val size: Int) : LocalChangesPaginator {
+  override fun page(list: List<LocalChange>) = list.chunked(size)
+}

engine/src/test/java/com/google/android/fhir/sync/upload/BundleUploaderTest.kt

@@ -46,7 +46,8 @@ class BundleUploaderTest {
           TestingUtils.BundleDataSource {
             Bundle().apply { type = Bundle.BundleType.TRANSACTIONRESPONSE }
           },
-          TransactionBundleGenerator.getDefault()
+          TransactionBundleGenerator.getDefault(),
+          LocalChangesPaginator.DEFAULT
         )
         .upload(localChanges)
         .toList()
@@ -69,7 +70,8 @@ class BundleUploaderTest {
               )
             }
           },
-          TransactionBundleGenerator.getDefault()
+          TransactionBundleGenerator.getDefault(),
+          LocalChangesPaginator.DEFAULT
         )
         .upload(localChanges)
         .toList()
@@ -83,7 +85,8 @@ class BundleUploaderTest {
     val result =
       BundleUploader(
           TestingUtils.BundleDataSource { throw ConnectException("Failed to connect to server.") },
-          TransactionBundleGenerator.getDefault()
+          TransactionBundleGenerator.getDefault(),
+          LocalChangesPaginator.DEFAULT
         )
         .upload(localChanges)
         .toList()

engine/src/test/java/com/google/android/fhir/sync/upload/LocalChangesPaginatorTest.kt

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.android.fhir.sync.upload
+
+import com.google.android.fhir.LocalChange
+import com.google.android.fhir.db.impl.dao.LocalChangeToken
+import com.google.common.truth.Truth.assertThat
+import org.hl7.fhir.r4.model.ResourceType
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+
+@RunWith(RobolectricTestRunner::class)
+class LocalChangesPaginatorTest {
+
+  @Test
+  fun `SizeSplitter split should return 3 lists in order`() {
+    val changes =
+      listOf(
+        LocalChange(
+          ResourceType.Patient.name,
+          "patient-001",
+          type = LocalChange.Type.INSERT,
+          payload = "{}",
+          token = LocalChangeToken(listOf(1L))
+        ),
+        LocalChange(
+          ResourceType.Patient.name,
+          "patient-002",
+          type = LocalChange.Type.INSERT,
+          payload = "{}",
+          token = LocalChangeToken(listOf(2L))
+        ),
+        LocalChange(
+          ResourceType.Patient.name,
+          "patient-003",
+          type = LocalChange.Type.INSERT,
+          payload = "{}",
+          token = LocalChangeToken(listOf(3L))
+        ),
+        LocalChange(
+          ResourceType.Patient.name,
+          "patient-004",
+          type = LocalChange.Type.INSERT,
+          payload = "{}",
+          token = LocalChangeToken(listOf(4L))
+        ),
+        LocalChange(
+          ResourceType.Patient.name,
+          "patient-005",
+          type = LocalChange.Type.INSERT,
+          payload = "{}",
+          token = LocalChangeToken(listOf(5L))
+        )
+      )
+
+    val result = SizeBasedLocalChangesPaginator(2).page(changes)
+    assertThat(result).hasSize(3)
+    assertThat(result[0].map { it.resourceId }).containsExactly("patient-001", "patient-002")
+    assertThat(result[1].map { it.resourceId }).containsExactly("patient-003", "patient-004")
+    assertThat(result[2].map { it.resourceId }).containsExactly("patient-005")
+  }
+
+  @Test
+  fun `SizeSplitter split should return empty lists when input its empty`() {
+    val result = SizeBasedLocalChangesPaginator(2).page(emptyList())
+    assertThat(result).hasSize(0)
+  }
+}