fix(storage): Prevent duplicate observer registration in TransferWorkerObserver to mitigate OOM issues (#2988)

Co-authored-by: Matt Creaser <[email protected]>

Author: onseok

aws-storage-s3/src/main/java/com/amplifyframework/storage/s3/transfer/TransferWorkerObserver.kt

@@ -25,6 +25,7 @@ import com.amplifyframework.storage.TransferState
 import com.amplifyframework.storage.s3.AWSS3StoragePlugin
 import com.amplifyframework.storage.s3.TransferOperations
 import com.amplifyframework.storage.s3.transfer.worker.BaseTransferWorker
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executors
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -49,6 +50,8 @@ internal class TransferWorkerObserver private constructor(
             AWSS3StoragePlugin.AWS_S3_STORAGE_LOG_NAMESPACE.format(this::class.java.simpleName)
         )
 
+    private val observedTags = ConcurrentHashMap.newKeySet<String>()
+
     init {
         attachObserverForPendingTransfer()
     }
@@ -195,13 +198,15 @@ internal class TransferWorkerObserver private constructor(
 
     private suspend fun attachObserver(tag: String) {
         withContext(Dispatchers.Main) {
+            if (!observedTags.add(tag)) return@withContext
             val liveData = workManager.getWorkInfosByTagLiveData(tag)
             liveData.observeForever(this@TransferWorkerObserver)
         }
     }
 
     private suspend fun removeObserver(tag: String) {
         withContext(Dispatchers.Main) {
+            if (!observedTags.remove(tag)) return@withContext
             workManager.getWorkInfosByTagLiveData(tag)
                 .removeObserver(this@TransferWorkerObserver)
         }