Add opt-in CoroutineDispatcher api for controlling Picasso's internal…

… thread (#2402)

picasso-paparazzi-sample/src/test/java/com/example/picasso/paparazzi/PicassoPaparazziTest.kt

@@ -26,6 +26,7 @@ import com.squareup.picasso3.RequestHandler
 import com.squareup.picasso3.layoutlib.LayoutlibExecutorService
 import org.junit.Rule
 import org.junit.Test
+import kotlinx.coroutines.Dispatchers
 
 class PicassoPaparazziTest {
   @get:Rule val paparazzi = Paparazzi()
@@ -35,6 +36,7 @@ class PicassoPaparazziTest {
     val picasso = Picasso.Builder(paparazzi.context)
       .callFactory { throw AssertionError() } // Removes network
       .executor(LayoutlibExecutorService())
+      .dispatcher(Dispatchers.Main)
       .addRequestHandler(FakeRequestHandler())
       .build()
 

picasso/api/picasso.api

@@ -107,6 +107,7 @@ public final class com/squareup/picasso3/Picasso$Builder {
 	public final fun callFactory (Lokhttp3/Call$Factory;)Lcom/squareup/picasso3/Picasso$Builder;
 	public final fun client (Lokhttp3/OkHttpClient;)Lcom/squareup/picasso3/Picasso$Builder;
 	public final fun defaultBitmapConfig (Landroid/graphics/Bitmap$Config;)Lcom/squareup/picasso3/Picasso$Builder;
+	public final fun dispatcher (Lkotlinx/coroutines/CoroutineDispatcher;)Lcom/squareup/picasso3/Picasso$Builder;
 	public final fun executor (Ljava/util/concurrent/ExecutorService;)Lcom/squareup/picasso3/Picasso$Builder;
 	public final fun indicatorsEnabled (Z)Lcom/squareup/picasso3/Picasso$Builder;
 	public final fun listener (Lcom/squareup/picasso3/Picasso$Listener;)Lcom/squareup/picasso3/Picasso$Builder;

picasso/src/main/java/com/squareup/picasso3/InternalCoroutineDispatcher.kt

@@ -0,0 +1,112 @@
+/*
+ * Copyright (C) 2023 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.squareup.picasso3
+
+import android.content.Context
+import android.net.NetworkInfo
+import android.os.Handler
+import java.util.concurrent.ExecutorService
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+
+internal class InternalCoroutineDispatcher internal constructor(
+  context: Context,
+  service: ExecutorService,
+  mainThreadHandler: Handler,
+  cache: PlatformLruCache,
+  val picassoDispatcher: CoroutineDispatcher
+) : Dispatcher(context, service, mainThreadHandler, cache) {
+
+  @OptIn(ExperimentalCoroutinesApi::class)
+  private val scope = CoroutineScope(picassoDispatcher.limitedParallelism(1))
+
+  override fun shutdown() {
+    super.shutdown()
+    scope.cancel()
+  }
+
+  override fun dispatchSubmit(action: Action) {
+    scope.launch {
+      performSubmit(action)
+    }
+  }
+
+  override fun dispatchCancel(action: Action) {
+    scope.launch {
+      performCancel(action)
+    }
+  }
+
+  override fun dispatchPauseTag(tag: Any) {
+    scope.launch {
+      performPauseTag(tag)
+    }
+  }
+
+  override fun dispatchResumeTag(tag: Any) {
+    scope.launch {
+      performResumeTag(tag)
+    }
+  }
+
+  override fun dispatchComplete(hunter: BitmapHunter) {
+    scope.launch {
+      performComplete(hunter)
+    }
+  }
+
+  override fun dispatchRetry(hunter: BitmapHunter) {
+    scope.launch {
+      delay(RETRY_DELAY)
+      performRetry(hunter)
+    }
+  }
+
+  override fun dispatchFailed(hunter: BitmapHunter) {
+    scope.launch {
+      performError(hunter)
+    }
+  }
+
+  override fun dispatchNetworkStateChange(info: NetworkInfo) {
+    scope.launch {
+      performNetworkStateChange(info)
+    }
+  }
+
+  override fun dispatchAirplaneModeChange(airplaneMode: Boolean) {
+    scope.launch {
+      performAirplaneModeChange(airplaneMode)
+    }
+  }
+
+  override fun dispatchCompleteMain(hunter: BitmapHunter) {
+    scope.launch(Dispatchers.Main) {
+      performCompleteMain(hunter)
+    }
+  }
+
+  override fun dispatchBatchResumeMain(batch: MutableList<Action>) {
+    scope.launch(Dispatchers.Main) {
+      performBatchResumeMain(batch)
+    }
+  }
+}

picasso/src/main/java/com/squareup/picasso3/Picasso.kt

@@ -51,6 +51,7 @@ import java.io.File
 import java.io.IOException
 import java.util.WeakHashMap
 import java.util.concurrent.ExecutorService
+import kotlinx.coroutines.CoroutineDispatcher
 
 /**
  * Image downloading, transformation, and caching manager.
@@ -537,6 +538,7 @@ class Picasso internal constructor(
     private val context: Context
     private var callFactory: Call.Factory? = null
     private var service: ExecutorService? = null
+    private var picassoDispatcher: CoroutineDispatcher? = null
     private var cache: PlatformLruCache? = null
     private var listener: Listener? = null
     private val requestTransformers = mutableListOf<RequestTransformer>()
@@ -555,6 +557,7 @@ class Picasso internal constructor(
       context = picasso.context
       callFactory = picasso.callFactory
       service = picasso.dispatcher.service
+      picassoDispatcher = (picasso.dispatcher as? InternalCoroutineDispatcher)?.picassoDispatcher
       cache = picasso.cache
       listener = picasso.listener
       requestTransformers += picasso.requestTransformers
@@ -647,6 +650,13 @@ class Picasso internal constructor(
       loggingEnabled = enabled
     }
 
+    /**
+     * Sets the dispatcher used internally for synchronizing access internally
+     */
+    fun dispatcher(picassoDispatcher: CoroutineDispatcher) = apply {
+      this.picassoDispatcher = picassoDispatcher
+    }
+
     /** Create the [Picasso] instance. */
     fun build(): Picasso {
       var unsharedCache: okhttp3.Cache? = null
@@ -665,7 +675,9 @@ class Picasso internal constructor(
         service = PicassoExecutorService()
       }
 
-      val dispatcher = HandlerDispatcher(context, service!!, HANDLER, cache!!)
+      val dispatcher = picassoDispatcher?.let {
+        InternalCoroutineDispatcher(context, service!!, HANDLER, cache!!, it)
+      } ?: HandlerDispatcher(context, service!!, HANDLER, cache!!)
 
       return Picasso(
         context, dispatcher, callFactory!!, unsharedCache, cache!!, listener,

picasso/src/test/java/com/squareup/picasso3/DispatcherTest.kt

@@ -66,9 +66,28 @@ import java.lang.Exception
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Future
 import java.util.concurrent.FutureTask
+import kotlinx.coroutines.Dispatchers
 
 @RunWith(RobolectricTestRunner::class)
-class DispatcherTest {
+class HandlerDispatcherTest : DispatcherTest() {
+  override fun createDispatcher(
+    context: Context,
+    service: ExecutorService,
+    cache: PlatformLruCache
+  ) = HandlerDispatcher(context, service, Handler(getMainLooper()), cache)
+}
+
+@RunWith(RobolectricTestRunner::class)
+class CoroutineDispatcherTest : DispatcherTest() {
+  override fun createDispatcher(
+    context: Context,
+    service: ExecutorService,
+    cache: PlatformLruCache
+  ) = InternalCoroutineDispatcher(context, service, Handler(getMainLooper()), cache, Dispatchers.Main)
+}
+
+abstract class DispatcherTest {
+
   @Mock lateinit var context: Context
 
   @Mock lateinit var connectivityManager: ConnectivityManager
@@ -610,9 +629,11 @@ class DispatcherTest {
     `when`(context.checkCallingOrSelfPermission(anyString())).thenReturn(
       if (scansNetworkChanges) PERMISSION_GRANTED else PERMISSION_DENIED
     )
-    return HandlerDispatcher(context, service, Handler(getMainLooper()), cache)
+    return createDispatcher(context, service, cache)
   }
 
+  internal abstract fun createDispatcher(context: Context, service: ExecutorService, cache: PlatformLruCache): Dispatcher
+
   private fun noopAction(data: Request, onComplete: () -> Unit = { }): Action {
     return object : Action(picasso, data) {
       override fun complete(result: RequestHandler.Result) = onComplete()