decoupling consumer/producer in ChangeManager

Author: DoomsCoder

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

@@ -40,12 +40,60 @@ 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.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+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 {
+    private val scope = CoroutineScope(Dispatchers.Main + SupervisorJob())
+
+    private val changeFlow =
+        MutableSharedFlow<Pair<OpChanges, Any?>>(
+            extraBufferCapacity = 64,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST,
+        )
+
+    init {
+        scope.launch {
+            changeFlow.collect { (changes, handler) ->
+                notifySubscribers(changes, handler)
+            }
+        }
+    }
+
+    fun <T : Any> publish(
+        changes: T,
+        handler: Any? = null,
+    ) {
+        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 {
@@ -156,7 +204,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

@@ -17,6 +17,7 @@ package com.ichi2.anki.observability
 
 import anki.collection.OpChanges
 import com.ichi2.testutils.JvmTest
+import kotlinx.coroutines.test.advanceUntilIdle
 import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.equalTo
 import org.hamcrest.Matchers.greaterThan
@@ -25,6 +26,7 @@ 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 +109,52 @@ class ChangeManagerTest {
         }
     }
 }
+
+@RunWith(RobolectricTestRunner::class)
+class ChangeManagerPublishTest : JvmTest() {
+    @After
+    override fun tearDown() {
+        super.tearDown()
+        ChangeManager.clearSubscribers()
+    }
+
+    @Test
+    fun `publish notifies multiple subscribers asynchronously`() =
+        runTest {
+            ShadowLooper.idleMainLooper()
+
+            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)
+
+            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))
+        }
+}