feat: decouple consumer/producer in ChangeManager

Primarily designed for the API, where `suspend` methods are not feasible
 in ContentProviders.

Add a synchronous 'publish' method, and use a Channel to decouple the
producer and consumer.

Fixes 21102
Fixes part of 20907

Assisted-by: Claude Opus 4.7 - self-review
Co-authored-by: David Allison <62114487+david-allison@users.noreply.github.com>

Author: 2 people

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

@@ -39,6 +39,8 @@ import anki.collection.OpChangesWithId
 import anki.collection.opChanges
 import anki.import_export.ImportResponse
 import com.ichi2.anki.common.crashreporting.CrashReportService
+import com.ichi2.anki.observability.ChangeManager.publish
+import com.ichi2.anki.observability.ChangeManager.toOpChanges
 import com.ichi2.anki.utils.ext.ifNotZero
 import org.jetbrains.annotations.Contract
 import timber.log.Timber
@@ -69,6 +71,30 @@ object ChangeManager {
 
     val subscriberCount get() = subscribers.size
 
+    /**
+     * Delivers changes to [subscribers][ChangeManager.subscribers] off the caller's thread.
+     *
+     * @see publish
+     */
+    private val publisher = OpChangesPublisher { changes, handler -> notifySubscribers(changes, handler) }
+
+    /**
+     * Notifies `subscribers` of `changes` without blocking the caller.
+     *
+     * Unlike `notifySubscribers`, the caller **does not** wait for subscribers to complete.
+     *
+     * This may be called from any thread.
+     *
+     * @param changes a raw/wrapped backend [OpChanges] response (see [ChangeManager.toOpChanges])
+     * @param handler the initiator of the change, or `null`. See [Subscriber.opExecuted]
+     */
+    fun <T : Any> publish(
+        changes: T,
+        handler: Any? = null,
+    ) {
+        publisher.publish(changes.toOpChanges(), handler)
+    }
+
     /**
      * Subscribes to changes.
      *
@@ -143,7 +169,9 @@ object ChangeManager {
     }
 
     @VisibleForTesting(otherwise = VisibleForTesting.NONE)
+    @Synchronized
     fun resetForTesting() {
+        publisher.reset()
         subscribers.size.ifNotZero { size -> Timber.d("clearing %d subscribers", size) }
         subscribers.clear()
     }
@@ -152,19 +180,26 @@ object ChangeManager {
         changes: T,
         initiator: Any?,
     ) {
-        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}'")
-            }
-        notifySubscribers(opChanges, initiator)
+        notifySubscribers(changes.toOpChanges(), initiator)
     }
 
+    /**
+     * Extracts an [OpChanges] from a backend response.
+     *
+     * @throws NotImplementedError if the receiver is not a backend OpChanges type.
+     */
+    // TODO: See if we can add a marker interface
+    private fun <T : Any> T.toOpChanges(): OpChanges =
+        when (this) {
+            is OpChanges -> this
+            is OpChangesWithCount -> changes
+            is OpChangesWithId -> changes
+            is OpChangesAfterUndo -> changes
+            is OpChangesOnly -> changes
+            is ImportResponse -> changes
+            else -> TODO("unhandled change type of class '${this::class}'")
+        }
+
     fun notifySubscribersAllValuesChanged(handler: Any? = null) {
         notifySubscribers(ALL, handler)
     }

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

@@ -0,0 +1,77 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package com.ichi2.anki.observability
+
+import anki.collection.OpChanges
+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
+
+/**
+ * Provides non-blocking delivery of [OpChanges] to the [ChangeManager] subscriber registry.
+ *
+ * Use this when `suspend` is not feasible (e.g. `ContentProvider`).
+ *
+ * [publish] queues a change onto [changeChannel] and returns immediately.
+ *
+ * @param deliver forwards a change to subscribers; invoked on [Dispatchers.Main].
+ */
+internal class OpChangesPublisher(
+    private val deliver: (changes: OpChanges, handler: Any?) -> Unit,
+) {
+    private data class ChangeEvent(
+        val changes: OpChanges,
+        val handler: Any?,
+    )
+
+    /** CoroutineScope which handles the coroutine consuming [changeChannel]. */
+    private var scope: CoroutineScope? = null
+    private val changeChannel =
+        Channel<ChangeEvent>(
+            capacity = Channel.UNLIMITED,
+        )
+
+    /** Ensures that [changeChannel] is being processed. */
+    @Synchronized
+    private fun ensureCollector() {
+        if (scope != null) return
+        scope =
+            CoroutineScope(Dispatchers.Main + SupervisorJob()).also { s ->
+                s.launch {
+                    for (event in changeChannel) {
+                        deliver(event.changes, event.handler)
+                    }
+                }
+            }
+    }
+
+    /**
+     * Queues [changes] for non-blocking delivery to subscribers.
+     *
+     * May be called on any thread.
+     *
+     * @see ChangeManager.publish
+     */
+    fun publish(
+        changes: OpChanges,
+        handler: Any?,
+    ) {
+        ensureCollector()
+        changeChannel.trySend(ChangeEvent(changes, handler))
+    }
+
+    /** Cancels the collector and discards any queued, undelivered changes. */
+    @Synchronized
+    fun reset() {
+        scope?.cancel()
+        scope = null
+        changeChannel.drain()
+    }
+}
+
+private fun <T> Channel<T>.drain() {
+    while (this.tryReceive().isSuccess) { /* intentionally empty */ }
+}

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

@@ -15,21 +15,29 @@
  */
 package com.ichi2.anki.observability
 
+import androidx.test.ext.junit.runners.AndroidJUnit4
 import anki.collection.OpChanges
+import com.ichi2.testutils.EmptyApplication
 import com.ichi2.testutils.JvmTest
+import com.ichi2.testutils.subscriberChangeCounter
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.setMain
 import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.equalTo
 import org.hamcrest.Matchers.greaterThan
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
 import kotlin.reflect.KProperty1
 import kotlin.reflect.full.memberProperties
 import kotlin.reflect.javaType
 import kotlin.reflect.jvm.isAccessible
 
-@RunWith(RobolectricTestRunner::class)
+// this cannot yet use `EmptyApplication` - `CrashReportService` dependency
+@RunWith(AndroidJUnit4::class)
 class ChangeManagerExceptionHandlingTest : JvmTest() {
     @Test
     fun `all subscribers notified even if one throws exception`() {
@@ -100,3 +108,26 @@ class ChangeManagerTest {
         }
     }
 }
+
+@RunWith(AndroidJUnit4::class)
+@Config(application = EmptyApplication::class)
+class ChangeManagerPublishTest : JvmTest() {
+    @Test
+    fun `publish notifies multiple subscribers asynchronously`() =
+        runTest {
+            // queue coroutine execution on Dispatchers.Main, to test 'not notified synchronously'
+            Dispatchers.setMain(StandardTestDispatcher(testScheduler))
+
+            val counter1 = subscriberChangeCounter()
+            val counter2 = subscriberChangeCounter()
+
+            ChangeManager.publish(ChangeManager.ALL)
+
+            assertThat("not notified synchronously by publish", counter1.hasChanges, equalTo(false))
+
+            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
+}