Adjust LCP downloads

Author: mickael-menu

readium/lcp/src/main/java/org/readium/r2/lcp/LcpDownloadsRepository.kt

@@ -9,67 +9,60 @@ package org.readium.r2.lcp
 import android.content.Context
 import java.io.File
 import java.util.LinkedList
-import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.MainScope
 import kotlinx.coroutines.async
-import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.plus
 import kotlinx.coroutines.withContext
 import org.json.JSONObject
+import org.readium.r2.shared.util.CoroutineQueue
 
 internal class LcpDownloadsRepository(
     context: Context
 ) {
-    private val coroutineScope: CoroutineScope =
-        MainScope()
+    private val queue = CoroutineQueue()
 
     private val storageDir: Deferred<File> =
-        coroutineScope.async {
+        queue.scope.async {
             withContext(Dispatchers.IO) {
                 File(context.noBackupFilesDir, LcpDownloadsRepository::class.qualifiedName!!)
                     .also { if (!it.exists()) it.mkdirs() }
             }
         }
 
     private val storageFile: Deferred<File> =
-        coroutineScope.async {
+        queue.scope.async {
             withContext(Dispatchers.IO) {
                 File(storageDir.await(), "licenses.json")
                     .also { if (!it.exists()) { it.writeText("{}", Charsets.UTF_8) } }
             }
         }
 
     private val snapshot: Deferred<MutableMap<String, JSONObject>> =
-        coroutineScope.async {
+        queue.scope.async {
             readSnapshot().toMutableMap()
         }
 
     fun addDownload(id: String, license: JSONObject) {
-        coroutineScope.launch {
+        queue.scope.launch {
             val snapshotCompleted = snapshot.await()
             snapshotCompleted[id] = license
             writeSnapshot(snapshotCompleted)
         }
     }
 
     fun removeDownload(id: String) {
-        coroutineScope.launch {
+        queue.launch {
             val snapshotCompleted = snapshot.await()
             snapshotCompleted.remove(id)
             writeSnapshot(snapshotCompleted)
         }
     }
 
-    suspend fun retrieveLicense(id: String): JSONObject? {
-        val snapshot = snapshot.await()
-        while (coroutineScope.coroutineContext.job.children.toList().isNotEmpty()) {
-            coroutineScope.coroutineContext.job.children.forEach { it.join() }
+    suspend fun retrieveLicense(id: String): JSONObject? =
+        queue.await {
+            snapshot.await()[id]
         }
-        return snapshot[id]
-    }
 
     private suspend fun readSnapshot(): Map<String, JSONObject> {
         return withContext(Dispatchers.IO) {

readium/lcp/src/main/java/org/readium/r2/lcp/LcpPublicationRetriever.kt

@@ -85,7 +85,7 @@ public class LcpPublicationRetriever(
             downloadTitle,
             downloadDescription
         )
-        register(requestId, listener)
+        addListener(requestId, listener)
         return requestId
     }
 
@@ -99,10 +99,16 @@ public class LcpPublicationRetriever(
         requestId: RequestId,
         listener: Listener
     ) {
-        listeners.getOrPut(requestId) {
-            downloadManager.register(DownloadManager.RequestId(requestId.value), downloadListener)
-            mutableListOf()
-        }.add(listener)
+        addListener(
+            requestId,
+            listener,
+            onFirstListenerAdded = {
+                downloadManager.register(
+                    DownloadManager.RequestId(requestId.value),
+                    downloadListener
+                )
+            }
+        )
     }
 
     /**
@@ -137,6 +143,19 @@ public class LcpPublicationRetriever(
     private val listeners: MutableMap<RequestId, MutableList<Listener>> =
         mutableMapOf()
 
+    private fun addListener(
+        requestId: RequestId,
+        listener: Listener,
+        onFirstListenerAdded: () -> Unit = {}
+    ) {
+        listeners
+            .getOrPut(requestId) {
+                onFirstListenerAdded()
+                mutableListOf()
+            }
+            .add(listener)
+    }
+
     private fun fetchPublication(
         license: LicenseDocument,
         downloadTitle: String,

readium/shared/src/main/java/org/readium/r2/shared/util/CoroutineQueue.kt

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2023 Readium Foundation. All rights reserved.
+ * Use of this source code is governed by the BSD-style license
+ * available in the top-level LICENSE file of the project.
+ */
+
+package org.readium.r2.shared.util
+
+import kotlin.coroutines.resume
+import kotlinx.coroutines.CancellableContinuation
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.MainScope
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.suspendCancellableCoroutine
+import org.readium.r2.shared.InternalReadiumApi
+
+/**
+ * Executes coroutines in a sequential order (FIFO).
+ */
+@InternalReadiumApi
+public class CoroutineQueue(
+    public val scope: CoroutineScope = MainScope()
+) {
+    init {
+        scope.launch {
+            for (task in tasks) {
+                task()
+            }
+        }
+    }
+
+    /**
+     * Launches a coroutine in the queue.
+     */
+    public fun launch(task: suspend () -> Unit) {
+        tasks.trySendBlocking(Task(task)).getOrThrow()
+    }
+
+    /**
+     * Launches a coroutine in the queue, and waits for its result.
+     */
+    public suspend fun <T> await(task: suspend () -> T): T =
+        suspendCancellableCoroutine { cont ->
+            tasks.trySendBlocking(Task(task, cont)).getOrThrow()
+        }
+
+    /**
+     * Cancels all the coroutines in the queue.
+     */
+    public fun cancel(cause: CancellationException? = null) {
+        scope.cancel(cause)
+    }
+
+    private val tasks: Channel<Task<*>> = Channel(Channel.UNLIMITED)
+
+    private class Task<T>(
+        val task: suspend () -> T,
+        val continuation: CancellableContinuation<T>? = null
+    ) {
+        suspend operator fun invoke() {
+            val result = task()
+            continuation?.resume(result)
+        }
+    }
+}

readium/shared/src/main/java/org/readium/r2/shared/util/downloads/android/AndroidDownloadManager.kt

@@ -248,7 +248,7 @@ public class AndroidDownloadManager internal constructor(
                 maybeStopObservingProgress()
             }
             SystemDownloadManager.STATUS_RUNNING -> {
-                val expected = facade.expected
+                val expected = facade.expected?.takeIf { it > 0 }
                 listenersForId.forEach {
                     it.onDownloadProgressed(id, facade.downloadedSoFar, expected)
                 }