decoupling consumer/producer in ChangeManager

Author: DoomsCoder

AnkiDroid/src/main/java/com/ichi2/anki/observability/ChangeManager.kt

@@ -40,12 +40,88 @@ import anki.collection.opChanges
 import anki.import_export.ImportResponse
 import com.ichi2.anki.common.crashreporting.CrashReportService
 import com.ichi2.anki.utils.ext.ifNotZero
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.launch
 import org.jetbrains.annotations.Contract
 import timber.log.Timber
 import java.lang.ref.WeakReference
 import java.util.concurrent.CopyOnWriteArrayList
 
 object ChangeManager {
+    @get:VisibleForTesting
+    @set:VisibleForTesting
+    internal var dispatcher: CoroutineDispatcher = Dispatchers.Main
+        set(value) {
+            if (field == value) return
+            field = value
+
+            resetCollector()
+        }
+
+    private var scope: CoroutineScope? = null
+    private var collectorJob: Job? = null
+
+    private val changeFlow =
+        MutableSharedFlow<Pair<OpChanges, Any?>>(
+            extraBufferCapacity = 64,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST,
+        )
+
+    private fun ensureCollector() {
+        if (collectorJob?.isActive == true) return
+
+        val currentScope = CoroutineScope(dispatcher + SupervisorJob())
+        scope = currentScope
+        collectorJob =
+            currentScope.launch(start = CoroutineStart.UNDISPATCHED) {
+                changeFlow.collect { (changes, handler) ->
+                    notifySubscribers(changes, handler)
+                }
+            }
+    }
+
+    private fun resetCollector() {
+        collectorJob?.cancel()
+        collectorJob = null
+        scope?.cancel()
+        scope = null
+    }
+
+    fun <T : Any> publish(
+        changes: T,
+        handler: Any? = null,
+    ) {
+        ensureCollector()
+        val opChanges =
+            when (changes) {
+                is OpChanges -> changes
+                is OpChangesWithCount -> changes.changes
+                is OpChangesWithId -> changes.changes
+                is OpChangesAfterUndo -> changes.changes
+                is OpChangesOnly -> changes.changes
+                is ImportResponse -> changes.changes
+                else -> {
+                    Timber.e("unhandled change type: ${changes::class}")
+                    return
+                }
+            }
+
+        val success = changeFlow.tryEmit(Pair(opChanges, handler))
+        if (!success) {
+            scope?.launch {
+                changeFlow.emit(Pair(opChanges, handler))
+            }
+        }
+    }
+
     // do not make this a 'fun interface' - lambdas may immediately be GCed
     // due to the use of WeakReference
     interface Subscriber {
@@ -78,6 +154,7 @@ object ChangeManager {
         subscriber: Subscriber,
         owner: LifecycleOwner? = subscriber as? LifecycleOwner,
     ) {
+        ensureCollector()
         val weakRef = WeakReference(subscriber)
 
         if (owner == null) {
@@ -156,7 +233,10 @@ object ChangeManager {
                 is OpChangesAfterUndo -> changes.changes
                 is OpChangesOnly -> changes.changes
                 is ImportResponse -> changes.changes
-                else -> TODO("unhandled change type of class '${changes::class}'")
+                else -> {
+                    Timber.e("unhandled change type: ${changes::class}")
+                    return
+                }
             }
         notifySubscribers(opChanges, initiator)
     }

AnkiDroid/src/test/java/com/ichi2/anki/observability/ChangeManagerTest.kt

@@ -16,15 +16,24 @@
 package com.ichi2.anki.observability
 
 import anki.collection.OpChanges
+import com.ichi2.anki.RobolectricTest
 import com.ichi2.testutils.JvmTest
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.yield
 import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.equalTo
 import org.hamcrest.Matchers.greaterThan
 import org.junit.After
+import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
 import org.robolectric.RobolectricTestRunner
+import org.robolectric.shadows.ShadowLooper
 import kotlin.reflect.KProperty1
 import kotlin.reflect.full.memberProperties
 import kotlin.reflect.javaType
@@ -107,3 +116,62 @@ class ChangeManagerTest {
         }
     }
 }
+
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+class ChangeManagerPublishTest : RobolectricTest() {
+    private val testDispatcher = StandardTestDispatcher()
+
+    @Before
+    override fun setUp() {
+        super.setUp()
+        ChangeManager.dispatcher = testDispatcher
+        setupTestDispatcher(testDispatcher)
+    }
+
+    @After
+    override fun tearDown() {
+        super.tearDown()
+        ChangeManager.clearSubscribers()
+        ChangeManager.dispatcher = Dispatchers.Main
+    }
+
+    @Test
+    fun `publish notifies multiple subscribers asynchronously`() =
+        runTest(testDispatcher) {
+            var firstNotified = false
+            var secondNotified = false
+
+            val subscriber1 =
+                object : ChangeManager.Subscriber {
+                    override fun opExecuted(
+                        changes: OpChanges,
+                        handler: Any?,
+                    ) {
+                        firstNotified = true
+                    }
+                }
+
+            val subscriber2 =
+                object : ChangeManager.Subscriber {
+                    override fun opExecuted(
+                        changes: OpChanges,
+                        handler: Any?,
+                    ) {
+                        secondNotified = true
+                    }
+                }
+
+            ChangeManager.subscribe(subscriber1)
+            ChangeManager.subscribe(subscriber2)
+
+            runCurrent()
+            ChangeManager.publish(ChangeManager.ALL)
+
+            advanceUntilIdle()
+            ShadowLooper.idleMainLooper()
+
+            assertThat("First subscriber should be notified", firstNotified, equalTo(true))
+            assertThat("Second subscriber should be notified", secondNotified, equalTo(true))
+        }
+}