use CopyOnWriteArrayList to remove the synchronized blocks

Author: polbins

core/src/main/java/com/amplitude/core/platform/Mediator.kt

@@ -4,17 +4,16 @@ import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.events.GroupIdentifyEvent
 import com.amplitude.core.events.IdentifyEvent
 import com.amplitude.core.events.RevenueEvent
+import java.util.concurrent.CopyOnWriteArrayList
 
-internal class Mediator(private val plugins: MutableList<Plugin>) {
-    fun add(plugin: Plugin) = synchronized(plugins) {
+internal class Mediator(private val plugins: CopyOnWriteArrayList<Plugin>) {
+    fun add(plugin: Plugin) {
         plugins.add(plugin)
     }
 
-    fun remove(plugin: Plugin) = synchronized(plugins) {
-        plugins.removeAll { it === plugin }
-    }
+    fun remove(plugin: Plugin) = plugins.removeAll { it === plugin }
 
-    fun execute(event: BaseEvent): BaseEvent? = synchronized(plugins) {
+    fun execute(event: BaseEvent): BaseEvent? {
         var result: BaseEvent? = event
 
         plugins.forEach { plugin ->
@@ -23,7 +22,7 @@ internal class Mediator(private val plugins: MutableList<Plugin>) {
                 result = when (plugin) {
                     is DestinationPlugin -> {
                         try {
-                            plugin.process(eventToBeProcessed)
+                            plugin.process(result)
                         } catch (e: Exception) {
                             e.printStackTrace()
                             null
@@ -46,7 +45,7 @@ internal class Mediator(private val plugins: MutableList<Plugin>) {
         return result
     }
 
-    fun applyClosure(closure: (Plugin) -> Unit) = synchronized(plugins) {
+    fun applyClosure(closure: (Plugin) -> Unit) {
         plugins.forEach {
             closure(it)
         }

core/src/main/java/com/amplitude/core/platform/Timeline.kt

@@ -2,13 +2,14 @@ package com.amplitude.core.platform
 
 import com.amplitude.core.Amplitude
 import com.amplitude.core.events.BaseEvent
+import java.util.concurrent.CopyOnWriteArrayList
 
 open class Timeline {
     internal val plugins: Map<Plugin.Type, Mediator> = mapOf(
-        Plugin.Type.Before to Mediator(mutableListOf()),
-        Plugin.Type.Enrichment to Mediator(mutableListOf()),
-        Plugin.Type.Destination to Mediator(mutableListOf()),
-        Plugin.Type.Utility to Mediator(mutableListOf())
+        Plugin.Type.Before to Mediator(CopyOnWriteArrayList()),
+        Plugin.Type.Enrichment to Mediator(CopyOnWriteArrayList()),
+        Plugin.Type.Destination to Mediator(CopyOnWriteArrayList()),
+        Plugin.Type.Utility to Mediator(CopyOnWriteArrayList())
     )
     lateinit var amplitude: Amplitude
 

core/src/test/kotlin/com/amplitude/core/platform/MediatorTest.kt

@@ -0,0 +1,76 @@
+package com.amplitude.core.platform
+
+import com.amplitude.core.Amplitude
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Timeout
+import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import kotlin.concurrent.thread
+
+class MediatorTest {
+    private val mediator = Mediator(CopyOnWriteArrayList())
+
+    /**
+     * Fake [DestinationPlugin] that does work on [flush] for 1 second
+     */
+    private class FakeDestinationPlugin : DestinationPlugin() {
+        var workDone = false
+
+        override fun flush() {
+            super.flush()
+            println("$this start work ${System.currentTimeMillis()} ${Thread.currentThread().name}")
+            Thread.sleep(1_000)
+            println("$this end work ${System.currentTimeMillis()} ${Thread.currentThread().name}")
+            workDone = true
+        }
+    }
+
+    @Test
+    @Timeout(2, unit = TimeUnit.SECONDS)
+    fun `multiple threads that call flush on destination plugin`() {
+        val fakeDestinationPlugins = List(10) { FakeDestinationPlugin() }
+        fakeDestinationPlugins.forEach {
+            mediator.add(it)
+        }
+
+        // simulate 10 threads executing flush on 10 different [DestinationPlugin]s
+        val executor = Executors.newFixedThreadPool(10)
+        fakeDestinationPlugins.forEach {
+            executor.submit {
+                it.flush()
+            }
+        }
+        executor.shutdown()
+        executor.awaitTermination(2, TimeUnit.SECONDS)
+
+        assertTrue {
+            fakeDestinationPlugins.all { it.workDone }
+        }
+    }
+
+    @Test
+    @Timeout(2, unit = TimeUnit.SECONDS)
+    fun `two threads that call flush on destination plugin`() {
+        val fakeDestinationPlugins = List(2) { FakeDestinationPlugin() }
+        fakeDestinationPlugins.forEach {
+            mediator.add(it)
+        }
+
+        // simulate 2 threads executing flush on 2 different DestinationPlugins
+        val t1 = thread {
+            fakeDestinationPlugins[0].flush()
+        }
+        val t2 = thread {
+            fakeDestinationPlugins[1].flush()
+        }
+        t1.join()
+        t2.join()
+
+        assertTrue {
+            fakeDestinationPlugins.all { it.workDone }
+        }
+    }
+
+}