Prevent ClosedReceiveChannelException after shutdown (#2473)

square · May 1, 2024 · 2c110ab · 2c110ab
1 parent c2dd184
commit 2c110ab
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 10 deletions.
diff --git a/picasso/src/main/java/com/squareup/picasso3/InternalCoroutineDispatcher.kt b/picasso/src/main/java/com/squareup/picasso3/InternalCoroutineDispatcher.kt
@@ -27,15 +27,14 @@ import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 
-@OptIn(ExperimentalCoroutinesApi::class)
 internal class InternalCoroutineDispatcher internal constructor(
   context: Context,
   mainThreadHandler: Handler,
@@ -50,8 +49,8 @@ internal class InternalCoroutineDispatcher internal constructor(
   init {
     // Using a channel to enforce sequential access for this class' internal state
     scope.launch {
-      while (!channel.isClosedForReceive) {
-        channel.receive().invoke()
+      channel.receiveAsFlow().collect {
+        it.invoke()
       }
     }
   }

diff --git a/picasso/src/test/java/com/squareup/picasso3/InternalCoroutineDispatcherTest.kt b/picasso/src/test/java/com/squareup/picasso3/InternalCoroutineDispatcherTest.kt
@@ -41,6 +41,7 @@ import org.robolectric.Shadows
 import java.lang.Exception
 import java.lang.RuntimeException
 import kotlin.coroutines.CoroutineContext
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestDispatcher
@@ -77,6 +78,17 @@ class InternalCoroutineDispatcherTest {
     assertThat(action.completedResult).isNull()
   }
 
+  @Test fun shutdownPreventsFurtherChannelUse() {
+    val dispatcher = createDispatcher(true, backgroundContext = Dispatchers.IO)
+    val action = TestUtils.mockAction(picasso, TestUtils.URI_KEY_1, TestUtils.URI_1)
+    dispatcher.shutdown()
+
+    dispatcher.dispatchSubmit(action)
+
+    assertThat(dispatcher.isShutdown()).isEqualTo(true)
+    assertThat(action.completedResult).isNull()
+  }
+
   @Test fun shutdownUnregistersReceiver() {
     dispatcher.shutdown()
     Shadows.shadowOf(Looper.getMainLooper()).idle()
@@ -729,7 +741,8 @@ class InternalCoroutineDispatcherTest {
 
   private fun createDispatcher(
     scansNetworkChanges: Boolean = true,
-    mainContext: CoroutineContext? = null
+    mainContext: CoroutineContext? = null,
+    backgroundContext: CoroutineContext? = null
   ): InternalCoroutineDispatcher {
     Mockito.`when`(connectivityManager.activeNetworkInfo).thenReturn(
       if (scansNetworkChanges) Mockito.mock(NetworkInfo::class.java) else null
@@ -742,11 +755,11 @@ class InternalCoroutineDispatcherTest {
     testDispatcher = StandardTestDispatcher()
     picasso = TestUtils.mockPicasso(context).newBuilder().dispatchers(mainContext ?: testDispatcher, testDispatcher).build()
     return InternalCoroutineDispatcher(
-      context,
-      Handler(Looper.getMainLooper()),
-      cache,
-      mainContext ?: testDispatcher,
-      testDispatcher
+      context = context,
+      mainThreadHandler = Handler(Looper.getMainLooper()),
+      cache = cache,
+      mainContext = mainContext ?: testDispatcher,
+      backgroundContext = backgroundContext ?: testDispatcher
     )
   }