switching to Channel and improving test code

Author: DoomsCoder

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

@@ -40,57 +40,42 @@ 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.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 {
-    @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,
+    private val changeChannel =
+        Channel<Pair<OpChanges, Any?>>(
+            capacity = 64,
             onBufferOverflow = BufferOverflow.DROP_OLDEST,
         )
 
+    @Synchronized
     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)
+        if (scope != null) return
+        scope =
+            CoroutineScope(Dispatchers.Main + SupervisorJob()).also { s ->
+                s.launch {
+                    for (event in changeChannel) {
+                        notifySubscribers(event.first, event.second)
+                    }
                 }
             }
     }
 
-    private fun resetCollector() {
-        collectorJob?.cancel()
-        collectorJob = null
+    @VisibleForTesting(otherwise = VisibleForTesting.NONE)
+    @Synchronized
+    fun resetForTesting() {
         scope?.cancel()
         scope = null
     }
@@ -108,18 +93,10 @@ object ChangeManager {
                 is OpChangesAfterUndo -> changes.changes
                 is OpChangesOnly -> changes.changes
                 is ImportResponse -> changes.changes
-                else -> {
-                    Timber.e("unhandled change type: ${changes::class}")
-                    return
-                }
+                else -> TODO("unhandled change type of class '${changes::class}'")
             }
 
-        val success = changeFlow.tryEmit(Pair(opChanges, handler))
-        if (!success) {
-            scope?.launch {
-                changeFlow.emit(Pair(opChanges, handler))
-            }
-        }
+        changeChannel.trySend(Pair(opChanges, handler))
     }
 
     // do not make this a 'fun interface' - lambdas may immediately be GCed
@@ -233,10 +210,7 @@ object ChangeManager {
                 is OpChangesAfterUndo -> changes.changes
                 is OpChangesOnly -> changes.changes
                 is ImportResponse -> changes.changes
-                else -> {
-                    Timber.e("unhandled change type: ${changes::class}")
-                    return
-                }
+                else -> TODO("unhandled change type of class '${changes::class}'")
             }
         notifySubscribers(opChanges, initiator)
     }

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

@@ -18,17 +18,14 @@ package com.ichi2.anki.observability
 import anki.collection.OpChanges
 import com.ichi2.anki.RobolectricTest
 import com.ichi2.testutils.JvmTest
-import kotlinx.coroutines.Dispatchers
+import com.ichi2.testutils.subscriberChangeCounter
 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
@@ -120,58 +117,26 @@ 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
+        ChangeManager.resetForTesting()
     }
 
     @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)
+        runTest {
+            val counter1 = subscriberChangeCounter()
+            val counter2 = subscriberChangeCounter()
 
             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))
+            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

@@ -74,7 +74,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
@@ -99,3 +99,9 @@ private class ExtractOpHandler : ChangeManager.Subscriber {
         this.handler = handler
     }
 }
+
+internal fun subscriberChangeCounter(): ChangeCounter {
+    val counter = ChangeCounter()
+    ChangeManager.subscribe(counter)
+    return counter
+}