fix(api): Fix concurrent access to OperationsManager

Fixes concurrent access to the `OperationsManager` by using a `ConcurrentHashMap`.

commit-id:15796d77

Author: Dillon Nys

packages/api/amplify_api_android/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/AmplifyApi.kt

@@ -75,7 +75,8 @@ class AmplifyApi : FlutterPlugin, MethodCallHandler {
     private val handler = Handler(Looper.getMainLooper())
 
     override fun onAttachedToEngine(@NonNull flutterPluginBinding: FlutterPlugin.FlutterPluginBinding) {
-        graphqlSubscriptionStreamHandler = graphqlSubscriptionStreamHandler ?: GraphQLSubscriptionStreamHandler()
+        graphqlSubscriptionStreamHandler =
+            graphqlSubscriptionStreamHandler ?: GraphQLSubscriptionStreamHandler()
         channel = MethodChannel(flutterPluginBinding.binaryMessenger, "com.amazonaws.amplify/api")
         channel!!.setMethodCallHandler(this)
         eventchannel = EventChannel(

packages/api/amplify_api_android/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/OperationsManager.kt

@@ -16,12 +16,17 @@
 package com.amazonaws.amplify.amplify_api
 
 import com.amplifyframework.core.async.Cancelable
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import java.util.concurrent.ConcurrentHashMap
 
 class OperationsManager {
 
     companion object {
 
-        private var operationsMap: HashMap<String, Cancelable> = HashMap()
+        private val scope = CoroutineScope(Dispatchers.IO)
+        private val operationsMap: ConcurrentHashMap<String, Cancelable> = ConcurrentHashMap()
 
         fun containsOperation(cancelToken: String): Boolean {
             return operationsMap.containsKey(cancelToken)
@@ -31,15 +36,20 @@ class OperationsManager {
             operationsMap[cancelToken] = operation
         }
 
-        fun removeOperation(cancelToken: String) {
-            if (containsOperation(cancelToken)) {
+        fun removeOperation(cancelToken: String): Cancelable? {
+            return if (containsOperation(cancelToken)) {
                 operationsMap.remove(cancelToken)
-            }
+            } else null
         }
 
         fun cancelOperation(cancelToken: String) {
-            operationsMap[cancelToken]?.cancel()
-            removeOperation(cancelToken)
+            val operation = removeOperation(cancelToken)
+            // Perform actual cancellation on a background thread.
+            operation?.let {
+                scope.launch {
+                    it.cancel()
+                }
+            }
         }
     }
 }

packages/api/amplify_api_android/android/src/test/kotlin/com/amazonaws/amplify/amplify_api/AmplifyApiRestTest.kt

@@ -31,11 +31,7 @@ import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.mockito.ArgumentCaptor
-import org.mockito.Mockito.any
-import org.mockito.Mockito.doAnswer
-import org.mockito.Mockito.mock
-import org.mockito.Mockito.times
-import org.mockito.Mockito.verify
+import org.mockito.Mockito.*
 import org.robolectric.RobolectricTestRunner
 import java.lang.reflect.Field
 import java.lang.reflect.Modifier

packages/api/amplify_api_android/android/src/test/kotlin/com/amazonaws/amplify/amplify_api/OperationsManagerTests.kt

@@ -0,0 +1,55 @@
+/*
+ *
+ * Copyright 2022 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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.amazonaws.amplify.amplify_api
+
+import com.amplifyframework.core.async.Cancelable
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runBlockingTest
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.times
+import org.mockito.kotlin.verify
+import org.robolectric.RobolectricTestRunner
+
+@ExperimentalCoroutinesApi
+@RunWith(RobolectricTestRunner::class)
+class OperationsManagerTests {
+
+    @Test
+    fun testConcurrentCancellation() = runBlockingTest {
+        val cancelToken = "cancelToken"
+        val operation = mock<Cancelable>()
+        OperationsManager.addOperation(cancelToken, operation)
+
+        launch {
+            // Launch 10 coroutines and wait til they all complete
+            (0..10).map {
+                async(Dispatchers.IO) {
+                    OperationsManager.cancelOperation(cancelToken)
+                }
+            }.awaitAll()
+        }
+
+        verify(operation, times(1)).cancel()
+    }
+}