Merge pull request #2034 from OneSignal/fix/event-producer-thread-safe

[Fix] thread safety with EventProducer's fire events

Author: jkasten2

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/events/EventProducer.kt

@@ -45,7 +45,7 @@ open class EventProducer<THandler> : IEventNotifier<THandler> {
      * @param callback The callback will be invoked for each subscribed handler, allowing you to call the handler.
      */
     fun fire(callback: (THandler) -> Unit) {
-        val localList = subscribers.toList()
+        val localList = synchronized(subscribers) { subscribers.toList() }
         for (s in localList) {
             callback(s)
         }
@@ -60,7 +60,7 @@ open class EventProducer<THandler> : IEventNotifier<THandler> {
      */
     fun fireOnMain(callback: (THandler) -> Unit) {
         suspendifyOnMain {
-            val localList = subscribers.toList()
+            val localList = synchronized(subscribers) { subscribers.toList() }
             for (s in localList) {
                 callback(s)
             }
@@ -74,7 +74,7 @@ open class EventProducer<THandler> : IEventNotifier<THandler> {
      * @param callback The callback will be invoked for each subscribed handler, allowing you to call the handler.
      */
     suspend fun suspendingFire(callback: suspend (THandler) -> Unit) {
-        val localList = subscribers.toList()
+        val localList = synchronized(subscribers) { subscribers.toList() }
         for (s in localList) {
             callback(s)
         }
@@ -88,7 +88,7 @@ open class EventProducer<THandler> : IEventNotifier<THandler> {
      */
     suspend fun suspendingFireOnMain(callback: suspend (THandler) -> Unit) {
         withContext(Dispatchers.Main) {
-            val localList = subscribers.toList()
+            val localList = synchronized(subscribers) { subscribers.toList() }
             for (s in localList) {
                 callback(s)
             }

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/EventProducerTest.kt

@@ -0,0 +1,50 @@
+package com.onesignal.common
+
+import com.onesignal.common.events.EventProducer
+import io.kotest.core.spec.style.FunSpec
+import kotlin.concurrent.thread
+
+class EventProducerTest : FunSpec({
+
+    fun modifyingSubscribersThread(eventProducer: EventProducer<Boolean>): Thread {
+        return thread(start = true) {
+            repeat(10_000) {
+                eventProducer.subscribe(true)
+                eventProducer.unsubscribe(true)
+            }
+        }
+    }
+
+    test("fire is thread safe") {
+        val eventProducer = EventProducer<Boolean>()
+        val modifyingSubscribersThread = modifyingSubscribersThread(eventProducer)
+
+        repeat(10_000) {
+            eventProducer.fire { }
+        }
+
+        modifyingSubscribersThread.join()
+    }
+
+    test("suspendingFire is thread safe") {
+        val eventProducer = EventProducer<Boolean>()
+        val modifyingSubscribersThread = modifyingSubscribersThread(eventProducer)
+
+        repeat(10_000) {
+            eventProducer.suspendingFire { }
+        }
+
+        modifyingSubscribersThread.join()
+    }
+
+    test("hasSubscribers is thread safe") {
+        val eventProducer = EventProducer<Boolean>()
+        val modifyingSubscribersThread = modifyingSubscribersThread(eventProducer)
+
+        repeat(10_000) {
+            eventProducer.hasSubscribers
+        }
+
+        modifyingSubscribersThread.join()
+    }
+})