fix(storage): Fix rare edge case of transfer status update crash (#3006)

Author: tylerjroach

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

@@ -36,11 +36,10 @@ internal class TransferStatusUpdater(
             AWSS3StoragePlugin.AWS_S3_STORAGE_LOG_NAMESPACE.format(this::class.java.simpleName)
         )
     private val mainHandler = Handler(Looper.getMainLooper())
-    private val transferStatusListenerMap: MutableMap<Int, MutableList<TransferListener>> by lazy {
-        ConcurrentHashMap()
-    }
-    private val transferWorkInfoIdMap: MutableMap<String, Int> by lazy { ConcurrentHashMap() }
-    private val multiPartTransferStatusListener: MutableMap<Int, MultiPartUploadTaskListener> by lazy {
+    private val transferStatusListenerMap:
+        ConcurrentHashMap<Int, ConcurrentHashMap.KeySetView<TransferListener, Boolean>> by lazy { ConcurrentHashMap() }
+    private val transferWorkInfoIdMap: ConcurrentHashMap<String, Int> by lazy { ConcurrentHashMap() }
+    private val multiPartTransferStatusListener: ConcurrentHashMap<Int, MultiPartUploadTaskListener> by lazy {
         ConcurrentHashMap()
     }
     val activeTransferMap = object : AbstractMutableMap<Int, TransferRecord>() {
@@ -135,9 +134,9 @@ internal class TransferStatusUpdater(
             transferDB.updateBytesTransferred(transferRecordId, bytesCurrent, bytesTotal)
         }
         if (notifyListener) {
-            transferStatusListenerMap[transferRecordId]?.forEach {
+            transferStatusListenerMap[transferRecordId]?.forEach { listener ->
                 mainHandler.post {
-                    it.onProgressChanged(
+                    listener.onProgressChanged(
                         transferRecordId,
                         bytesCurrent,
                         bytesTotal
@@ -149,8 +148,8 @@ internal class TransferStatusUpdater(
 
     @Synchronized
     fun updateOnError(transferRecordId: Int, exception: Exception) {
-        transferStatusListenerMap[transferRecordId]?.forEach {
-            mainHandler.post { it.onError(transferRecordId, exception) }
+        transferStatusListenerMap[transferRecordId]?.forEach { listener ->
+            mainHandler.post { listener.onError(transferRecordId, exception) }
         }
     }
 
@@ -162,13 +161,11 @@ internal class TransferStatusUpdater(
 
     @Synchronized
     fun registerListener(transferRecordId: Int, transferListener: TransferListener) {
-        transferStatusListenerMap[transferRecordId]?.let {
-            val transferListener = transferListener
-            if (!it.contains(transferListener)) {
-                it.add(transferListener)
+        transferStatusListenerMap[transferRecordId]?.add(transferListener) ?: run {
+            val transferRecordMap = ConcurrentHashMap.newKeySet<TransferListener>().apply {
+                add(transferListener)
             }
-        } ?: run {
-            transferStatusListenerMap[transferRecordId] = mutableListOf(transferListener)
+            transferStatusListenerMap[transferRecordId] = transferRecordMap
         }
     }