Implement ConcurrentRequestStrategy to support awaiting in-flight requests.

coil-network-core/api/android/coil-network-core.api

@@ -31,6 +31,15 @@ public final class coil3/network/CacheStrategy$WriteResult {
 public final class coil3/network/CacheStrategy$WriteResult$Companion {
 }
 
+public abstract interface class coil3/network/ConcurrentRequestStrategy {
+	public static final field Companion Lcoil3/network/ConcurrentRequestStrategy$Companion;
+	public static final field UNCOORDINATED Lcoil3/network/ConcurrentRequestStrategy;
+	public abstract fun apply (Ljava/lang/String;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
+public final class coil3/network/ConcurrentRequestStrategy$Companion {
+}
+
 public abstract interface class coil3/network/ConnectivityChecker {
 	public static final field Companion Lcoil3/network/ConnectivityChecker$Companion;
 	public static final field ONLINE Lcoil3/network/ConnectivityChecker;
@@ -44,6 +53,11 @@ public final class coil3/network/ConnectivityCheckerKt {
 	public static final fun ConnectivityChecker (Landroid/content/Context;)Lcoil3/network/ConnectivityChecker;
 }
 
+public final class coil3/network/DeDupeConcurrentRequestStrategy : coil3/network/ConcurrentRequestStrategy {
+	public fun <init> ()V
+	public fun apply (Ljava/lang/String;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
 public final class coil3/network/HttpException : java/lang/RuntimeException {
 	public fun <init> (Lcoil3/network/NetworkResponse;)V
 	public final fun getResponse ()Lcoil3/network/NetworkResponse;
@@ -74,13 +88,16 @@ public final class coil3/network/NetworkClientKt {
 }
 
 public final class coil3/network/NetworkFetcher : coil3/fetch/Fetcher {
-	public fun <init> (Ljava/lang/String;Lcoil3/request/Options;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lcoil3/network/ConnectivityChecker;)V
+	public synthetic fun <init> (Ljava/lang/String;Lcoil3/request/Options;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lcoil3/network/ConnectivityChecker;)V
+	public fun <init> (Ljava/lang/String;Lcoil3/request/Options;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;)V
 	public fun fetch (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
 public final class coil3/network/NetworkFetcher$Factory : coil3/fetch/Fetcher$Factory {
-	public fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;)V
+	public synthetic fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;)V
 	public synthetic fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function0;)V
+	public synthetic fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function0;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun create (Lcoil3/Uri;Lcoil3/request/Options;Lcoil3/ImageLoader;)Lcoil3/fetch/Fetcher;
 	public synthetic fun create (Ljava/lang/Object;Lcoil3/request/Options;Lcoil3/ImageLoader;)Lcoil3/fetch/Fetcher;
 }

coil-network-core/api/coil-network-core.klib.api

@@ -55,6 +55,15 @@ abstract interface coil3.network/CacheStrategy { // coil3.network/CacheStrategy|
     }
 }
 
+abstract interface coil3.network/ConcurrentRequestStrategy { // coil3.network/ConcurrentRequestStrategy|null[0]
+    abstract suspend fun apply(kotlin/String, kotlin.coroutines/SuspendFunction0<coil3.fetch/FetchResult>): coil3.fetch/FetchResult // coil3.network/ConcurrentRequestStrategy.apply|apply(kotlin.String;kotlin.coroutines.SuspendFunction0<coil3.fetch.FetchResult>){}[0]
+
+    final object Companion { // coil3.network/ConcurrentRequestStrategy.Companion|null[0]
+        final val UNCOORDINATED // coil3.network/ConcurrentRequestStrategy.Companion.UNCOORDINATED|{}UNCOORDINATED[0]
+            final fun <get-UNCOORDINATED>(): coil3.network/ConcurrentRequestStrategy // coil3.network/ConcurrentRequestStrategy.Companion.UNCOORDINATED.<get-UNCOORDINATED>|<get-UNCOORDINATED>(){}[0]
+    }
+}
+
 abstract interface coil3.network/NetworkClient { // coil3.network/NetworkClient|null[0]
     abstract suspend fun <#A1: kotlin/Any?> executeRequest(coil3.network/NetworkRequest, kotlin.coroutines/SuspendFunction1<coil3.network/NetworkResponse, #A1>): #A1 // coil3.network/NetworkClient.executeRequest|executeRequest(coil3.network.NetworkRequest;kotlin.coroutines.SuspendFunction1<coil3.network.NetworkResponse,0:0>){0§<kotlin.Any?>}[0]
 }
@@ -68,6 +77,12 @@ abstract interface coil3.network/NetworkResponseBody : kotlin/AutoCloseable { //
     abstract suspend fun writeTo(okio/FileSystem, okio/Path) // coil3.network/NetworkResponseBody.writeTo|writeTo(okio.FileSystem;okio.Path){}[0]
 }
 
+final class coil3.network/DeDupeConcurrentRequestStrategy : coil3.network/ConcurrentRequestStrategy { // coil3.network/DeDupeConcurrentRequestStrategy|null[0]
+    constructor <init>() // coil3.network/DeDupeConcurrentRequestStrategy.<init>|<init>(){}[0]
+
+    final suspend fun apply(kotlin/String, kotlin.coroutines/SuspendFunction0<coil3.fetch/FetchResult>): coil3.fetch/FetchResult // coil3.network/DeDupeConcurrentRequestStrategy.apply|apply(kotlin.String;kotlin.coroutines.SuspendFunction0<coil3.fetch.FetchResult>){}[0]
+}
+
 final class coil3.network/HttpException : kotlin/RuntimeException { // coil3.network/HttpException|null[0]
     constructor <init>(coil3.network/NetworkResponse) // coil3.network/HttpException.<init>|<init>(coil3.network.NetworkResponse){}[0]
 
@@ -77,11 +92,13 @@ final class coil3.network/HttpException : kotlin/RuntimeException { // coil3.net
 
 final class coil3.network/NetworkFetcher : coil3.fetch/Fetcher { // coil3.network/NetworkFetcher|null[0]
     constructor <init>(kotlin/String, coil3.request/Options, kotlin/Lazy<coil3.network/NetworkClient>, kotlin/Lazy<coil3.disk/DiskCache?>, kotlin/Lazy<coil3.network/CacheStrategy>, coil3.network/ConnectivityChecker) // coil3.network/NetworkFetcher.<init>|<init>(kotlin.String;coil3.request.Options;kotlin.Lazy<coil3.network.NetworkClient>;kotlin.Lazy<coil3.disk.DiskCache?>;kotlin.Lazy<coil3.network.CacheStrategy>;coil3.network.ConnectivityChecker){}[0]
+    constructor <init>(kotlin/String, coil3.request/Options, kotlin/Lazy<coil3.network/NetworkClient>, kotlin/Lazy<coil3.disk/DiskCache?>, kotlin/Lazy<coil3.network/CacheStrategy>, kotlin/Lazy<coil3.network/ConnectivityChecker>, kotlin/Lazy<coil3.network/ConcurrentRequestStrategy>) // coil3.network/NetworkFetcher.<init>|<init>(kotlin.String;coil3.request.Options;kotlin.Lazy<coil3.network.NetworkClient>;kotlin.Lazy<coil3.disk.DiskCache?>;kotlin.Lazy<coil3.network.CacheStrategy>;kotlin.Lazy<coil3.network.ConnectivityChecker>;kotlin.Lazy<coil3.network.ConcurrentRequestStrategy>){}[0]
 
     final suspend fun fetch(): coil3.fetch/FetchResult // coil3.network/NetworkFetcher.fetch|fetch(){}[0]
 
     final class Factory : coil3.fetch/Fetcher.Factory<coil3/Uri> { // coil3.network/NetworkFetcher.Factory|null[0]
         constructor <init>(kotlin/Function0<coil3.network/NetworkClient>, kotlin/Function0<coil3.network/CacheStrategy> = ..., kotlin/Function1<coil3/PlatformContext, coil3.network/ConnectivityChecker> = ...) // coil3.network/NetworkFetcher.Factory.<init>|<init>(kotlin.Function0<coil3.network.NetworkClient>;kotlin.Function0<coil3.network.CacheStrategy>;kotlin.Function1<coil3.PlatformContext,coil3.network.ConnectivityChecker>){}[0]
+        constructor <init>(kotlin/Function0<coil3.network/NetworkClient>, kotlin/Function0<coil3.network/CacheStrategy> = ..., kotlin/Function1<coil3/PlatformContext, coil3.network/ConnectivityChecker> = ..., kotlin/Function0<coil3.network/ConcurrentRequestStrategy> = ...) // coil3.network/NetworkFetcher.Factory.<init>|<init>(kotlin.Function0<coil3.network.NetworkClient>;kotlin.Function0<coil3.network.CacheStrategy>;kotlin.Function1<coil3.PlatformContext,coil3.network.ConnectivityChecker>;kotlin.Function0<coil3.network.ConcurrentRequestStrategy>){}[0]
 
         final fun create(coil3/Uri, coil3.request/Options, coil3/ImageLoader): coil3.fetch/Fetcher? // coil3.network/NetworkFetcher.Factory.create|create(coil3.Uri;coil3.request.Options;coil3.ImageLoader){}[0]
     }

coil-network-core/api/jvm/coil-network-core.api

@@ -31,6 +31,15 @@ public final class coil3/network/CacheStrategy$WriteResult {
 public final class coil3/network/CacheStrategy$WriteResult$Companion {
 }
 
+public abstract interface class coil3/network/ConcurrentRequestStrategy {
+	public static final field Companion Lcoil3/network/ConcurrentRequestStrategy$Companion;
+	public static final field UNCOORDINATED Lcoil3/network/ConcurrentRequestStrategy;
+	public abstract fun apply (Ljava/lang/String;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
+public final class coil3/network/ConcurrentRequestStrategy$Companion {
+}
+
 public abstract interface class coil3/network/ConnectivityChecker {
 	public static final field Companion Lcoil3/network/ConnectivityChecker$Companion;
 	public static final field ONLINE Lcoil3/network/ConnectivityChecker;
@@ -44,6 +53,11 @@ public final class coil3/network/ConnectivityCheckerKt {
 	public static final fun ConnectivityChecker (Lcoil3/PlatformContext;)Lcoil3/network/ConnectivityChecker;
 }
 
+public final class coil3/network/DeDupeConcurrentRequestStrategy : coil3/network/ConcurrentRequestStrategy {
+	public fun <init> ()V
+	public fun apply (Ljava/lang/String;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
 public final class coil3/network/HttpException : java/lang/RuntimeException {
 	public fun <init> (Lcoil3/network/NetworkResponse;)V
 	public final fun getResponse ()Lcoil3/network/NetworkResponse;
@@ -74,13 +88,16 @@ public final class coil3/network/NetworkClientKt {
 }
 
 public final class coil3/network/NetworkFetcher : coil3/fetch/Fetcher {
-	public fun <init> (Ljava/lang/String;Lcoil3/request/Options;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lcoil3/network/ConnectivityChecker;)V
+	public synthetic fun <init> (Ljava/lang/String;Lcoil3/request/Options;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lcoil3/network/ConnectivityChecker;)V
+	public fun <init> (Ljava/lang/String;Lcoil3/request/Options;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;Lkotlin/Lazy;)V
 	public fun fetch (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
 public final class coil3/network/NetworkFetcher$Factory : coil3/fetch/Fetcher$Factory {
-	public fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;)V
+	public synthetic fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;)V
 	public synthetic fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function0;)V
+	public synthetic fun <init> (Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function0;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun create (Lcoil3/Uri;Lcoil3/request/Options;Lcoil3/ImageLoader;)Lcoil3/fetch/Fetcher;
 	public synthetic fun create (Ljava/lang/Object;Lcoil3/request/Options;Lcoil3/ImageLoader;)Lcoil3/fetch/Fetcher;
 }

coil-network-core/src/commonMain/kotlin/coil3/network/ConcurrentRequestStrategy.kt

@@ -0,0 +1,102 @@
+package coil3.network
+
+import coil3.annotation.ExperimentalCoilApi
+import coil3.fetch.FetchResult
+import kotlin.jvm.JvmField
+import kotlinx.atomicfu.locks.SynchronizedObject
+import kotlinx.atomicfu.locks.synchronized
+import kotlinx.coroutines.channels.Channel
+
+/**
+ * Coordinates concurrent requests for the same key.
+ *
+ * Implementations can reduce duplicate work by running `block` once and making
+ * other callers wait for that result.
+ */
+@ExperimentalCoilApi
+interface ConcurrentRequestStrategy {
+    suspend fun apply(key: String, block: suspend () -> FetchResult): FetchResult
+
+    companion object {
+        /** Runs `block` immediately with no request coordination. */
+        @JvmField val UNCOORDINATED: ConcurrentRequestStrategy = UncoordinatedConcurrentRequestStrategy()
+    }
+}
+
+private class UncoordinatedConcurrentRequestStrategy : ConcurrentRequestStrategy {
+    override suspend fun apply(
+        key: String,
+        block: suspend () -> FetchResult,
+    ): FetchResult = block()
+}
+
+/**
+ * De-duplicates concurrent requests for the same key.
+ *
+ * The first caller executes `block`. If it succeeds, all waiters are released
+ * so they can continue (for example, by reading from cache). If it fails or is
+ * canceled, one waiter is resumed to retry `block`.
+ */
+@ExperimentalCoilApi
+class DeDupeConcurrentRequestStrategy : ConcurrentRequestStrategy {
+    private val concurrentRequests = mutableMapOf<String, Request>()
+    private val lock = SynchronizedObject()
+
+    override suspend fun apply(
+        key: String,
+        block: suspend () -> FetchResult,
+    ): FetchResult {
+        var shouldWait = true
+        val request = synchronized(lock) {
+            concurrentRequests.getOrPut(key) {
+                shouldWait = false
+                Request()
+            }
+        }.acquire()
+
+        if (shouldWait) {
+            request.channel.receiveCatching()
+        }
+
+        try {
+            return block().also {
+                request.markSucceeded()
+            }
+        } catch (e: Exception) {
+            request.channel.trySend(Unit)
+            throw e
+        } finally {
+            request.release {
+                synchronized(lock) {
+                    concurrentRequests -= key
+                }
+            }
+        }
+    }
+
+    private class Request {
+        val channel = Channel<Unit>(Channel.UNLIMITED)
+
+        private val lock = SynchronizedObject()
+        private var hasSucceeded = false
+        private var isClosed = false
+        private var observerCount = 0
+
+        fun acquire(): Request = synchronized(lock) {
+            observerCount++
+            this
+        }
+
+        fun markSucceeded() = synchronized(lock) {
+            hasSucceeded = true
+        }
+
+        fun release(cleanup: () -> Unit) = synchronized(lock) {
+            if ((--observerCount <= 0 || hasSucceeded) && !isClosed) {
+                channel.close()
+                cleanup()
+                isClosed = true
+            }
+        }
+    }
+}

coil-network-core/src/commonMain/kotlin/coil3/network/NetworkFetcher.kt

@@ -35,10 +35,33 @@ class NetworkFetcher(
     private val networkClient: Lazy<NetworkClient>,
     private val diskCache: Lazy<DiskCache?>,
     private val cacheStrategy: Lazy<CacheStrategy>,
-    private val connectivityChecker: ConnectivityChecker,
+    private val connectivityChecker: Lazy<ConnectivityChecker>,
+    private val concurrentRequestStrategy: Lazy<ConcurrentRequestStrategy>,
 ) : Fetcher {
 
+    @Deprecated("Kept for binary compatibility.", level = DeprecationLevel.HIDDEN)
+    constructor(
+        url: String,
+        options: Options,
+        networkClient: Lazy<NetworkClient>,
+        diskCache: Lazy<DiskCache?>,
+        cacheStrategy: Lazy<CacheStrategy>,
+        connectivityChecker: ConnectivityChecker,
+    ) : this(
+        url = url,
+        options = options,
+        networkClient = networkClient,
+        diskCache = diskCache,
+        cacheStrategy = cacheStrategy,
+        connectivityChecker = lazyOf(connectivityChecker),
+        concurrentRequestStrategy = lazyOf(ConcurrentRequestStrategy.UNCOORDINATED),
+    )
+
     override suspend fun fetch(): FetchResult {
+        return concurrentRequestStrategy.value.apply(diskCacheKey, ::doFetch)
+    }
+
+    private suspend fun doFetch(): FetchResult {
         var snapshot = readFromDiskCache()
         try {
             // Fast path: fetch the image from the disk cache without performing a network request.
@@ -183,7 +206,7 @@ class NetworkFetcher(
     private fun newRequest(): NetworkRequest {
         val headers = options.httpHeaders.newBuilder()
         val diskRead = options.diskCachePolicy.readEnabled
-        val networkRead = options.networkCachePolicy.readEnabled && connectivityChecker.isOnline()
+        val networkRead = options.networkCachePolicy.readEnabled && connectivityChecker.value.isOnline()
         when {
             !networkRead && diskRead -> {
                 headers[CACHE_CONTROL] = "only-if-cached, max-stale=2147483647"
@@ -265,10 +288,25 @@ class NetworkFetcher(
         networkClient: () -> NetworkClient,
         cacheStrategy: () -> CacheStrategy = { CacheStrategy.DEFAULT },
         connectivityChecker: (PlatformContext) -> ConnectivityChecker = ::ConnectivityChecker,
+        concurrentRequestStrategy: () -> ConcurrentRequestStrategy = { ConcurrentRequestStrategy.UNCOORDINATED },
     ) : Fetcher.Factory<Uri> {
+
+        @Deprecated("Kept for binary compatibility.", level = DeprecationLevel.HIDDEN)
+        constructor(
+            networkClient: () -> NetworkClient,
+            cacheStrategy: () -> CacheStrategy = { CacheStrategy.DEFAULT },
+            connectivityChecker: (PlatformContext) -> ConnectivityChecker = ::ConnectivityChecker,
+        ) : this(
+            networkClient = networkClient,
+            cacheStrategy = cacheStrategy,
+            connectivityChecker = connectivityChecker,
+            concurrentRequestStrategy = { ConcurrentRequestStrategy.UNCOORDINATED },
+        )
+
         private val networkClientLazy = lazy(networkClient)
         private val cacheStrategyLazy = lazy(cacheStrategy)
         private val connectivityCheckerLazy = singleParameterLazy(connectivityChecker)
+        private val concurrentRequestStrategyLazy = lazy(concurrentRequestStrategy)
 
         override fun create(
             data: Uri,
@@ -282,7 +320,8 @@ class NetworkFetcher(
                 networkClient = networkClientLazy,
                 diskCache = lazy { imageLoader.diskCache },
                 cacheStrategy = cacheStrategyLazy,
-                connectivityChecker = connectivityCheckerLazy.get(options.context),
+                connectivityChecker = lazyOf(connectivityCheckerLazy.get(options.context)),
+                concurrentRequestStrategy = concurrentRequestStrategyLazy,
             )
         }