feat: decouple consumer/producer in ChangeManager

Use a Channel to decouple the two, and a synchronous 'publish' method.

Fixes 21102
Fixes part of 20907

Co-authored-by: David Allison <62114487+david-allison@users.noreply.github.com>

Author: DoomsCoder

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

@@ -40,12 +40,56 @@ 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.cancel
+import kotlinx.coroutines.channels.Channel
+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 var scope: CoroutineScope? = null
+    private val changeChannel =
+        Channel<Pair<OpChanges, Any?>>(
+            capacity = Channel.UNLIMITED,
+        )
+
+    @Synchronized
+    private fun ensureCollector() {
+        if (scope != null) return
+        scope =
+            CoroutineScope(Dispatchers.Main + SupervisorJob()).also { s ->
+                s.launch {
+                    for (event in changeChannel) {
+                        notifySubscribers(event.first, event.second)
+                    }
+                }
+            }
+    }
+
+    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 -> TODO("unhandled change type of class '${changes::class}'")
+            }
+
+        changeChannel.trySend(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 +122,7 @@ object ChangeManager {
         subscriber: Subscriber,
         owner: LifecycleOwner? = subscriber as? LifecycleOwner,
     ) {
+        ensureCollector()
         val weakRef = WeakReference(subscriber)
 
         if (owner == null) {
@@ -139,7 +184,11 @@ object ChangeManager {
     }
 
     @VisibleForTesting(otherwise = VisibleForTesting.NONE)
+    @Synchronized
     fun resetForTesting() {
+        scope?.cancel()
+        scope = null
+        changeChannel.drain()
         subscribers.size.ifNotZero { size -> Timber.d("clearing %d subscribers", size) }
         subscribers.clear()
     }
@@ -189,3 +238,7 @@ object ChangeManager {
 private fun <T> MutableList<WeakReference<T>>.removeWeakReferences(block: (T?) -> Boolean) {
     this.removeAll { block(it.get()) }
 }
+
+private fun <T> Channel<T>.drain() {
+    while (this.tryReceive().isSuccess) { /* intentionally empty */ }
+}

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

@@ -17,6 +17,9 @@ package com.ichi2.anki.observability
 
 import anki.collection.OpChanges
 import com.ichi2.testutils.JvmTest
+import com.ichi2.testutils.subscriberChangeCounter
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
 import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.equalTo
 import org.hamcrest.Matchers.greaterThan
@@ -100,3 +103,21 @@ class ChangeManagerTest {
         }
     }
 }
+
+class ChangeManagerPublishTest : JvmTest() {
+    @Test
+    fun `publish notifies multiple subscribers asynchronously`() =
+        runTest {
+            val counter1 = subscriberChangeCounter()
+            val counter2 = subscriberChangeCounter()
+
+            runCurrent()
+
+            ChangeManager.publish(ChangeManager.ALL)
+
+            advanceUntilIdle()
+
+            assertThat("First subscriber should be notified", counter1.hasChanges, equalTo(true))
+            assertThat("Second subscriber should be notified", counter2.hasChanges, equalTo(true))
+        }
+}

AnkiDroid/src/test/java/com/ichi2/testutils/TestChangeSubscriber.kt

@@ -60,7 +60,7 @@ suspend fun ensureOpWithHandler(
 }
 
 // used to ensure a strong reference to the subscription is held
-private class ChangeCounter : ChangeManager.Subscriber {
+internal class ChangeCounter : ChangeManager.Subscriber {
     private var changes = 0
     val changeCount get() = changes
     val hasChanges get() = changes > 0
@@ -85,3 +85,14 @@ private class ExtractOpHandler : ChangeManager.Subscriber {
         this.handler = handler
     }
 }
+
+/**
+ * Produces a [ChangeCounter] which is subscribed to [ChangeManager].
+ *
+ * Query the result via [.changeCount][ChangeCounter.changeCount] or [.hasChanges][ChangeCounter.hasChanges]
+ */
+internal fun subscriberChangeCounter(): ChangeCounter {
+    val counter = ChangeCounter()
+    ChangeManager.subscribe(counter)
+    return counter
+}