refactor: Replace DownloadQueue with Flow (#301)

* refactor: Replace DownloadQueue with Flow

* fix: Remove DownloadQueue leftover

* fix: Download progress not progressing

* chore: Remove unnecessary downloadStatusChange trigger

Already handled by MainActivity

* fix: Chapter download state stuck in CHECKED

* fix: Chapter download state stuck in QUEUE on deletion

* fix: A regression, download progress not progressing

* refactor: Remove rx usage

* docs: Sync changelog

CHANGELOG.md

@@ -15,6 +15,11 @@ The format is simplified version of [Keep a Changelog](https://keepachangelog.co
 - Fix slow chapter load
 - Fix chapter bookmark state is not persistent
 
+## Other
+- Refactor downloader
+  - Replace RxJava usage with Kotlin coroutines
+  - Replace DownloadQueue with Flow to hopefully fix ConcurrentModificationException entirely
+
 ## [1.9.2]
 
 ### Changes

app/src/main/java/eu/kanade/tachiyomi/data/download/DownloadJob.kt

@@ -22,8 +22,9 @@ import eu.kanade.tachiyomi.util.system.workManager
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.asSharedFlow
+import kotlinx.coroutines.flow.map
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 import yokai.i18n.MR
@@ -39,7 +40,7 @@ class DownloadJob(val context: Context, workerParams: WorkerParameters) : Corout
     private val preferences: PreferencesHelper = Injekt.get()
 
     override suspend fun getForegroundInfo(): ForegroundInfo {
-        val firstDL = downloadManager.queue.firstOrNull()
+        val firstDL = downloadManager.queueState.value.firstOrNull()
         val notification = DownloadNotifier(context).setPlaceholder(firstDL).build()
         val id = Notifications.ID_DOWNLOAD_CHAPTER
         return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
@@ -70,7 +71,6 @@ class DownloadJob(val context: Context, workerParams: WorkerParameters) : Corout
         } catch (_: CancellationException) {
             Result.success()
         } finally {
-            callListeners(false, downloadManager)
             if (runExtJobAfter) {
                 ExtensionUpdateJob.runJobAgain(applicationContext, NetworkType.CONNECTED)
             }
@@ -96,12 +96,6 @@ class DownloadJob(val context: Context, workerParams: WorkerParameters) : Corout
         private const val TAG = "Downloader"
         private const val START_EXT_JOB_AFTER = "StartExtJobAfter"
 
-        private val downloadChannel = MutableSharedFlow<Boolean>(
-            extraBufferCapacity = 1,
-            onBufferOverflow = BufferOverflow.DROP_OLDEST,
-        )
-        val downloadFlow = downloadChannel.asSharedFlow()
-
         fun start(context: Context, alsoStartExtJob: Boolean = false) {
             val request = OneTimeWorkRequestBuilder<DownloadJob>()
                 .addTag(TAG)
@@ -118,16 +112,17 @@ class DownloadJob(val context: Context, workerParams: WorkerParameters) : Corout
             context.workManager.cancelUniqueWork(TAG)
         }
 
-        fun callListeners(downloading: Boolean? = null, downloadManager: DownloadManager? = null) {
-            val dManager by lazy { downloadManager ?: Injekt.get() }
-            downloadChannel.tryEmit(downloading ?: !dManager.isPaused())
-        }
-
         fun isRunning(context: Context): Boolean {
             return context.workManager
                 .getWorkInfosForUniqueWork(TAG)
                 .get()
                 .let { list -> list.count { it.state == WorkInfo.State.RUNNING } == 1 }
         }
+
+        fun isRunningFlow(context: Context): Flow<Boolean> {
+            return context.workManager
+                .getWorkInfosForUniqueWorkFlow(TAG)
+                .map { list -> list.count { it.state == WorkInfo.State.RUNNING } == 1 }
+        }
     }
 }

app/src/main/java/eu/kanade/tachiyomi/data/download/DownloadManager.kt

@@ -5,7 +5,6 @@ import co.touchlab.kermit.Logger
 import com.hippo.unifile.UniFile
 import eu.kanade.tachiyomi.data.database.models.Chapter
 import eu.kanade.tachiyomi.data.download.model.Download
-import eu.kanade.tachiyomi.data.download.model.DownloadQueue
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.domain.manga.models.Manga
 import eu.kanade.tachiyomi.source.Source
@@ -13,10 +12,14 @@ import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.util.system.ImageUtil
 import eu.kanade.tachiyomi.util.system.launchIO
-import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.flow.drop
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.merge
+import kotlinx.coroutines.flow.onStart
 import uy.kohesive.injekt.injectLazy
 import yokai.domain.download.DownloadPreferences
 import yokai.i18n.MR
@@ -65,8 +68,11 @@ class DownloadManager(val context: Context) {
     /**
      * Downloads queue, where the pending chapters are stored.
      */
-    val queue: DownloadQueue
+    val queueState
-        get() = downloader.queue
+        get() = downloader.queueState
+
+    val isDownloaderRunning
+        get() = DownloadJob.isRunningFlow(context)
 
     /**
      * Tells the downloader to begin downloads.
@@ -75,7 +81,6 @@ class DownloadManager(val context: Context) {
      */
     fun startDownloads(): Boolean {
         val hasStarted = downloader.start()
-        DownloadJob.callListeners(downloadManager = this)
         return hasStarted
     }
 
@@ -99,22 +104,21 @@ class DownloadManager(val context: Context) {
      *
      * @param isNotification value that determines if status is set (needed for view updates)
      */
-    fun clearQueue(isNotification: Boolean = false) {
+    fun clearQueue() {
-        deletePendingDownloads(*downloader.queue.toTypedArray())
+        deletePendingDownloads(*queueState.value.toTypedArray())
-        downloader.removeFromQueue(isNotification)
+        downloader.clearQueue()
-        DownloadJob.callListeners(false, this)
+        downloader.stop()
     }
 
     fun startDownloadNow(chapter: Chapter) {
-        val download = downloader.queue.find { it.chapter.id == chapter.id } ?: return
+        val download = queueState.value.find { it.chapter.id == chapter.id } ?: return
-        val queue = downloader.queue.toMutableList()
+        val queue = queueState.value.toMutableList()
         queue.remove(download)
         queue.add(0, download)
         reorderQueue(queue)
         if (isPaused()) {
             if (DownloadJob.isRunning(context)) {
                 downloader.start()
-                DownloadJob.callListeners(true, this)
             } else {
                 DownloadJob.start(context)
             }
@@ -127,24 +131,12 @@ class DownloadManager(val context: Context) {
      * @param downloads value to set the download queue to
      */
     fun reorderQueue(downloads: List<Download>) {
-        val wasPaused = isPaused()
+        downloader.updateQueue(downloads)
-        if (downloads.isEmpty()) {
-            DownloadJob.stop(context)
-            downloader.queue.clear()
-            return
-        }
-        downloader.pause()
-        downloader.queue.clear()
-        downloader.queue.addAll(downloads)
-        if (!wasPaused) {
-            downloader.start()
-            DownloadJob.callListeners(true, this)
-        }
     }
 
     fun isPaused() = !downloader.isRunning
 
-    fun hasQueue() = downloader.queue.isNotEmpty()
+    fun hasQueue() = queueState.value.isNotEmpty()
 
     /**
      * Tells the downloader to enqueue the given list of chapters.
@@ -164,10 +156,7 @@ class DownloadManager(val context: Context) {
      */
     fun addDownloadsToStartOfQueue(downloads: List<Download>) {
         if (downloads.isEmpty()) return
-        queue.toMutableList().apply {
+        reorderQueue(downloads + queueState.value)
-            addAll(0, downloads)
-            reorderQueue(this)
-        }
         if (!DownloadJob.isRunning(context)) DownloadJob.start(context)
     }
 
@@ -212,7 +201,7 @@ class DownloadManager(val context: Context) {
      * @param chapter the chapter to check.
      */
     fun getChapterDownloadOrNull(chapter: Chapter): Download? {
-        return downloader.queue
+        return queueState.value
             .firstOrNull { it.chapter.id == chapter.id && it.chapter.manga_id == chapter.manga_id }
     }
 
@@ -249,27 +238,15 @@ class DownloadManager(val context: Context) {
      * @param manga the manga of the chapters.
      * @param source the source of the chapters.
      */
-    @OptIn(DelicateCoroutinesApi::class)
     fun deleteChapters(chapters: List<Chapter>, manga: Manga, source: Source, force: Boolean = false) {
+        launchIO {
             val filteredChapters = if (force) chapters else getChaptersToDelete(chapters, manga)
-        GlobalScope.launch(Dispatchers.IO) {
-            val wasPaused = isPaused()
             if (filteredChapters.isEmpty()) {
-                return@launch
+                return@launchIO
             }
-            downloader.pause()
+
-            downloader.queue.remove(filteredChapters)
+            removeFromDownloadQueue(filteredChapters)
-            if (!wasPaused && downloader.queue.isNotEmpty()) {
+
-                downloader.start()
-                DownloadJob.callListeners(true)
-            } else if (downloader.queue.isEmpty() && DownloadJob.isRunning(context)) {
-                DownloadJob.callListeners(false)
-                DownloadJob.stop(context)
-            } else if (downloader.queue.isEmpty()) {
-                DownloadJob.callListeners(false)
-                downloader.stop()
-            }
-            queue.remove(filteredChapters)
             val chapterDirs =
                 provider.findChapterDirs(filteredChapters, manga, source) + provider.findTempChapterDirs(
                     filteredChapters,
@@ -278,10 +255,27 @@ class DownloadManager(val context: Context) {
                 )
             chapterDirs.forEach { it.delete() }
             cache.removeChapters(filteredChapters, manga)
+
             if (cache.getDownloadCount(manga, true) == 0) { // Delete manga directory if empty
                 chapterDirs.firstOrNull()?.parentFile?.delete()
             }
-            queue.updateListeners()
+        }
+    }
+
+    private fun removeFromDownloadQueue(chapters: List<Chapter>) {
+        val wasRunning = downloader.isRunning
+        if (wasRunning) {
+            downloader.pause()
+        }
+
+        downloader.removeFromQueue(chapters)
+
+        if (wasRunning) {
+            if (queueState.value.isEmpty()) {
+                downloader.stop()
+            } else if (queueState.value.isNotEmpty()) {
+                downloader.start()
+            }
         }
     }
 
@@ -345,9 +339,7 @@ class DownloadManager(val context: Context) {
     fun deleteManga(manga: Manga, source: Source, removeQueued: Boolean = true) {
         launchIO {
             if (removeQueued) {
-                downloader.removeFromQueue(manga, true)
+                downloader.removeFromQueue(manga)
-                queue.remove(manga)
-                queue.updateListeners()
             }
             provider.findMangaDir(manga, source)?.delete()
             cache.removeManga(manga)
@@ -418,9 +410,6 @@ class DownloadManager(val context: Context) {
         cache.forceRenewCache()
     }
 
-    fun addListener(listener: DownloadQueue.DownloadListener) = queue.addListener(listener)
-    fun removeListener(listener: DownloadQueue.DownloadListener) = queue.removeListener(listener)
-
     private fun getChaptersToDelete(chapters: List<Chapter>, manga: Manga): List<Chapter> {
         // Retrieve the categories that are set to exclude from being deleted on read
         return if (!preferences.removeBookmarkedChapters().get()) {
@@ -429,4 +418,33 @@ class DownloadManager(val context: Context) {
             chapters
         }
     }
+
+    fun statusFlow(): Flow<Download> = queueState
+        .flatMapLatest { downloads ->
+            downloads
+                .map { download ->
+                    download.statusFlow.drop(1).map { download }
+                }
+                .merge()
+        }
+        .onStart {
+            emitAll(
+                queueState.value.filter { download -> download.status == Download.State.DOWNLOADING }.asFlow(),
+            )
+        }
+
+    fun progressFlow(): Flow<Download> = queueState
+        .flatMapLatest { downloads ->
+            downloads
+                .map { download ->
+                    download.progressFlow.drop(1).map { download }
+                }
+                .merge()
+        }
+        .onStart {
+            emitAll(
+                queueState.value.filter { download -> download.status == Download.State.DOWNLOADING }
+                    .asFlow(),
+            )
+        }
 }

app/src/main/java/eu/kanade/tachiyomi/data/download/DownloadStore.kt

@@ -59,6 +59,12 @@ class DownloadStore(
         }
     }
 
+    fun removeAll(downloads: List<Download>) {
+        preferences.edit {
+            downloads.forEach { remove(getKey(it)) }
+        }
+    }
+
     /**
      * Removes all the downloads from the store.
      */

app/src/main/java/eu/kanade/tachiyomi/data/download/Downloader.kt

@@ -5,11 +5,9 @@ import android.os.Handler
 import android.os.Looper
 import co.touchlab.kermit.Logger
 import com.hippo.unifile.UniFile
-import com.jakewharton.rxrelay.PublishRelay
 import eu.kanade.tachiyomi.data.cache.ChapterCache
 import eu.kanade.tachiyomi.data.database.models.Chapter
 import eu.kanade.tachiyomi.data.download.model.Download
-import eu.kanade.tachiyomi.data.download.model.DownloadQueue
 import eu.kanade.tachiyomi.data.library.LibraryUpdateJob
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.domain.manga.models.Manga
@@ -32,22 +30,31 @@ import java.io.File
 import java.util.*
 import java.util.zip.*
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flatMapMerge
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.retryWhen
+import kotlinx.coroutines.flow.transformLatest
+import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.supervisorScope
 import nl.adaptivity.xmlutil.serialization.XML
 import okhttp3.Response
-import rx.Observable
-import rx.Subscription
-import rx.android.schedulers.AndroidSchedulers
-import rx.schedulers.Schedulers
 import uy.kohesive.injekt.injectLazy
 import yokai.core.archive.ZipWriter
 import yokai.core.metadata.COMIC_INFO_FILE
@@ -61,16 +68,7 @@ import yokai.util.lang.getString
 /**
  * This class is the one in charge of downloading chapters.
  *
- * Its [queue] contains the list of chapters to download. In order to download them, the downloader
+ * Its queue contains the list of chapters to download.
- * subscriptions must be running and the list of chapters must be sent to them by [downloadsRelay].
- *
- * The queue manipulation must be done in one thread (currently the main thread) to avoid unexpected
- * behavior, but it's safe to read it from multiple threads.
- *
- * @param context the application context.
- * @param provider the downloads directory provider.
- * @param cache the downloads cache, used to add the downloads to the cache after their completion.
- * @param sourceManager the source manager.
  */
 class Downloader(
     private val context: Context,
@@ -92,7 +90,8 @@ class Downloader(
     /**
      * Queue where active downloads are kept.
      */
-    val queue = DownloadQueue(store)
+    private val _queueState = MutableStateFlow<List<Download>>(emptyList())
+    val queueState = _queueState.asStateFlow()
 
     private val handler = Handler(Looper.getMainLooper())
 
@@ -101,21 +100,14 @@ class Downloader(
      */
     private val notifier by lazy { DownloadNotifier(context) }
 
-    /**
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
-     * Downloader subscription.
+    private var downloaderJob: Job? = null
-     */
-    private var subscription: Subscription? = null
-
-    /**
-     * Relay to send a list of downloads to the downloader.
-     */
-    private val downloadsRelay = PublishRelay.create<List<Download>>()
 
     /**
      * Whether the downloader is running.
      */
     val isRunning: Boolean
-        get() = subscription != null
+        get() = downloaderJob?.isActive ?: false
 
     /**
      * Whether the downloader is paused
@@ -126,8 +118,7 @@ class Downloader(
     init {
         launchNow {
             val chapters = async { store.restore() }
-            queue.addAll(chapters.await())
+            addAllToQueue(chapters.await())
-            DownloadJob.callListeners()
         }
     }
 
@@ -138,17 +129,17 @@ class Downloader(
      * @return true if the downloader is started, false otherwise.
      */
     fun start(): Boolean {
-        if (subscription != null || queue.isEmpty()) {
+        if (isRunning || queueState.value.isEmpty()) {
             return false
         }
-        initializeSubscription()
 
-        val pending = queue.filter { it.status != Download.State.DOWNLOADED }
+        val pending = queueState.value.filter { it.status != Download.State.DOWNLOADED }
         pending.forEach { if (it.status != Download.State.QUEUE) it.status = Download.State.QUEUE }
 
         isPaused = false
 
-        downloadsRelay.call(pending)
+        launchDownloaderJob()
+
         return pending.isNotEmpty()
     }
 
@@ -156,8 +147,8 @@ class Downloader(
      * Stops the downloader.
      */
     fun stop(reason: String? = null) {
-        destroySubscription()
+        cancelDownloaderJob()
-        queue
+        queueState.value
             .filter { it.status == Download.State.DOWNLOADING }
             .forEach { it.status = Download.State.ERROR }
 
@@ -166,104 +157,109 @@ class Downloader(
             return
         }
 
-        DownloadJob.stop(context)
+        if (isPaused && queueState.value.isNotEmpty()) {
-        if (isPaused && queue.isNotEmpty()) {
             handler.postDelayed({ notifier.onDownloadPaused() }, 150)
         } else {
             notifier.dismiss()
         }
-        DownloadJob.callListeners(false)
+
         isPaused = false
+
+        DownloadJob.stop(context)
     }
 
     /**
      * Pauses the downloader
      */
     fun pause() {
-        destroySubscription()
+        cancelDownloaderJob()
-        queue
+        queueState.value
             .filter { it.status == Download.State.DOWNLOADING }
             .forEach { it.status = Download.State.QUEUE }
         isPaused = true
     }
 
-    /**
+    fun clearQueue() {
-     * Removes everything from the queue.
+        cancelDownloaderJob()
-     *
-     * @param isNotification value that determines if status is set (needed for view updates)
-     */
-    fun removeFromQueue(isNotification: Boolean = false) {
-        destroySubscription()
 
-        // Needed to update the chapter view
+        internalClearQueue()
-        if (isNotification) {
-            queue
-                .filter { it.status == Download.State.QUEUE }
-                .forEach { it.status = Download.State.NOT_DOWNLOADED }
-        }
-        queue.clear()
         notifier.dismiss()
     }
 
-    /**
+    private fun launchDownloaderJob() {
-     * Removes everything from the queue for a certain manga
-     *
-     * @param isNotification value that determines if status is set (needed for view updates)
-     */
-    fun removeFromQueue(manga: Manga, isNotification: Boolean = false) {
-        // Needed to update the chapter view
-        if (isNotification) {
-            queue.filter { it.status == Download.State.QUEUE && it.manga.id == manga.id }
-                .forEach { it.status = Download.State.NOT_DOWNLOADED }
-        }
-        queue.remove(manga)
-        if (queue.isEmpty()) {
-            if (DownloadJob.isRunning(context)) DownloadJob.stop(context)
-            stop()
-        }
-        notifier.dismiss()
-    }
-
-    /**
-     * Prepares the subscriptions to start downloading.
-     */
-    private fun initializeSubscription() {
         if (isRunning) return
 
-        subscription = downloadsRelay.concatMapIterable { it }
+        downloaderJob = scope.launch {
-            // Concurrently download from 5 different sources
+            val activeDownloadsFlow = queueState.transformLatest { queue ->
-            .groupBy { it.source }
+                while (true) {
-            .flatMap(
+                    val activeDownloads = queue.asSequence()
-                { bySource ->
+                        // Ignore completed downloads, leave them in the queue
-                    bySource.concatMap { download ->
+                        .filter {
-                        Observable.fromCallable {
+                            val statusValue = it.status.value
-                            runBlocking { downloadChapter(download) }
+                            Download.State.NOT_DOWNLOADED.value <= statusValue && statusValue <= Download.State.DOWNLOADING.value
-                            download
-                        }.subscribeOn(Schedulers.io())
                         }
-                },
+                        .groupBy { it.source }
-                5,
+                        .toList()
-            )
+                        // Concurrently download from 5 different sources
-            .onBackpressureLatest()
+                        .take(5)
-            .observeOn(AndroidSchedulers.mainThread())
+                        .map { (_, downloads) -> downloads.first() }
-            .subscribe(
+                    emit(activeDownloads)
-                {
+
-                    completeDownload(it)
+                    if (activeDownloads.isEmpty()) break
-                },
+                    // Suspend until a download enters the ERROR state
-                { error ->
+                    val activeDownloadsErroredFlow =
-                    Logger.e(error)
+                        combine(activeDownloads.map(Download::statusFlow)) { states ->
-                    notifier.onError(error.message)
+                            states.contains(Download.State.ERROR)
+                        }.filter { it }
+                    activeDownloadsErroredFlow.first()
+                }
+            }.distinctUntilChanged()
+
+            // Use supervisorScope to cancel child jobs when the downloader job is cancelled
+            supervisorScope {
+                val downloadJobs = mutableMapOf<Download, Job>()
+
+                activeDownloadsFlow.collectLatest { activeDownloads ->
+                    val downloadJobsToStop = downloadJobs.filter { it.key !in activeDownloads }
+                    downloadJobsToStop.forEach { (download, job) ->
+                        job.cancel()
+                        downloadJobs.remove(download)
+                    }
+
+                    val downloadsToStart = activeDownloads.filter { it !in downloadJobs }
+                    downloadsToStart.forEach { download ->
+                        downloadJobs[download] = launchDownloadJob(download)
+                    }
+                }
+            }
+        }
+    }
+
+    private fun CoroutineScope.launchDownloadJob(download: Download) = launchIO {
+        try {
+            downloadChapter(download)
+
+            // Remove successful download from queue
+            if (download.status == Download.State.DOWNLOADED) {
+                removeFromQueue(download)
+            }
+            if (areAllDownloadsFinished()) {
                 stop()
-                },
+            }
-            )
+        } catch (e: Throwable) {
+            if (e is CancellationException) throw e
+            Logger.e(e)
+            notifier.onError(e.message)
+            stop()
+        }
     }
 
     /**
      * Destroys the downloader subscriptions.
      */
-    private fun destroySubscription() {
+    private fun cancelDownloaderJob() {
-        subscription?.unsubscribe()
+        downloaderJob?.cancel()
-        subscription = null
+        downloaderJob = null
     }
 
     /**
@@ -279,7 +275,7 @@ class Downloader(
         }
 
         val source = sourceManager.get(manga.source) as? HttpSource ?: return@launchIO
-        val wasEmpty = queue.isEmpty()
+        val wasEmpty = queueState.value.isEmpty()
         // Called in background thread, the operation can be slow with SAF.
         val chaptersWithoutDir = async {
             chapters
@@ -292,22 +288,17 @@ class Downloader(
         // Runs in main thread (synchronization needed).
         val chaptersToQueue = chaptersWithoutDir.await()
             // Filter out those already enqueued.
-            .filter { chapter -> queue.none { it.chapter.id == chapter.id } }
+            .filter { chapter -> queueState.value.none { it.chapter.id == chapter.id } }
             // Create a download for each one.
             .map { Download(source, manga, it) }
 
         if (chaptersToQueue.isNotEmpty()) {
-            queue.addAll(chaptersToQueue)
+            addAllToQueue(chaptersToQueue)
-
-            if (isRunning) {
-                // Send the list of downloads to the downloader.
-                downloadsRelay.call(chaptersToQueue)
-            }
 
             // Start downloader if needed
             if (autoStart && wasEmpty) {
-                val queuedDownloads = queue.count { it.source !is UnmeteredSource }
+                val queuedDownloads = queueState.value.count { it.source !is UnmeteredSource }
-                val maxDownloadsFromSource = queue
+                val maxDownloadsFromSource = queueState.value
                     .groupBy { it.source }
                     .filterKeys { it !is UnmeteredSource }
                     .maxOfOrNull { it.value.size } ?: 0
@@ -670,25 +661,86 @@ class Downloader(
         dir.createFile(COMIC_INFO_FILE)?.writeText(xml.encodeToString(ComicInfo.serializer(), comicInfo))
     }
 
-    /**
-     * Completes a download. This method is called in the main thread.
-     */
-    private fun completeDownload(download: Download) {
-        // Delete successful downloads from queue
-        if (download.status == Download.State.DOWNLOADED) {
-            // Remove downloaded chapter from queue
-            queue.remove(download)
-        }
-        if (areAllDownloadsFinished()) {
-            stop()
-        }
-    }
-
     /**
      * Returns true if all the queued downloads are in DOWNLOADED or ERROR state.
      */
     private fun areAllDownloadsFinished(): Boolean {
-        return queue.none { it.status <= Download.State.DOWNLOADING }
+        return queueState.value.none { it.status <= Download.State.DOWNLOADING }
+    }
+
+    private fun addAllToQueue(downloads: List<Download>) {
+        _queueState.update {
+            downloads.forEach { download ->
+                download.status = Download.State.QUEUE
+            }
+            store.addAll(downloads)
+            it + downloads
+        }
+    }
+
+    fun removeFromQueue(download: Download) {
+        _queueState.update {
+            store.remove(download)
+            if (download.status == Download.State.DOWNLOADING || download.status == Download.State.QUEUE) {
+                download.status = Download.State.NOT_DOWNLOADED
+            }
+            it - download
+        }
+    }
+
+    private inline fun removeFromQueueIf(predicate: (Download) -> Boolean) {
+        _queueState.update { queue ->
+            val downloads = queue.filter { predicate(it) }
+            store.removeAll(downloads)
+            downloads.forEach { download ->
+                if (download.status == Download.State.DOWNLOADING || download.status == Download.State.QUEUE) {
+                    download.status = Download.State.NOT_DOWNLOADED
+                }
+            }
+            queue - downloads
+        }
+    }
+
+    fun removeFromQueue(chapter: Chapter) {
+        removeFromQueueIf { it.chapter.id == chapter.id }
+    }
+
+    fun removeFromQueue(chapters: List<Chapter>) {
+        removeFromQueueIf { it.chapter.id in chapters.map { it.id } }
+    }
+
+    fun removeFromQueue(manga: Manga) {
+        removeFromQueueIf { it.manga.id == manga.id }
+    }
+
+    private fun internalClearQueue() {
+        _queueState.update {
+            it.forEach { download ->
+                if (download.status == Download.State.DOWNLOADING || download.status == Download.State.QUEUE) {
+                    download.status = Download.State.NOT_DOWNLOADED
+                }
+            }
+            store.clear()
+            emptyList()
+        }
+    }
+
+    fun updateQueue(downloads: List<Download>) {
+        val wasRunning = isRunning
+
+        if (downloads.isEmpty()) {
+            clearQueue()
+            DownloadJob.stop(context)
+            return
+        }
+
+        pause()
+        internalClearQueue()
+        addAllToQueue(downloads)
+
+        if (wasRunning) {
+            start()
+        }
     }
 
     companion object {

app/src/main/java/eu/kanade/tachiyomi/data/download/model/Download.kt

@@ -4,8 +4,15 @@ import eu.kanade.tachiyomi.data.database.models.Chapter
 import eu.kanade.tachiyomi.domain.manga.models.Manga
 import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.source.online.HttpSource
-import rx.subjects.PublishSubject
 import kotlin.math.roundToInt
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.debounce
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flow
 
 class Download(val source: HttpSource, val manga: Manga, val chapter: Chapter) {
 
@@ -17,17 +24,31 @@ class Download(val source: HttpSource, val manga: Manga, val chapter: Chapter) {
     val downloadedImages: Int
         get() = pages?.count { it.status == Page.State.READY } ?: 0
 
-    @Volatile @Transient
+    @Transient
-    var status: State = State.default
+    private val _statusFlow = MutableStateFlow(State.NOT_DOWNLOADED)
+
+    @Transient
+    val statusFlow = _statusFlow.asStateFlow()
+    var status: State
+        get() = _statusFlow.value
         set(status) {
-            field = status
+            _statusFlow.value = status
-            statusSubject?.onNext(this)
-            statusCallback?.invoke(this)
         }
 
-    @Transient private var statusSubject: PublishSubject<Download>? = null
+    @Transient
+    val progressFlow = flow {
+        if (pages == null) {
+            emit(0)
+            while (pages == null) {
+                delay(50)
+            }
+        }
 
-    @Transient private var statusCallback: ((Download) -> Unit)? = null
+        val progressFlows = pages!!.map(Page::progressFlow)
+        emitAll(combine(progressFlows) { it.average().roundToInt() })
+    }
+        .distinctUntilChanged()
+        .debounce(50)
 
     val pageProgress: Int
         get() {
@@ -41,21 +62,13 @@ class Download(val source: HttpSource, val manga: Manga, val chapter: Chapter) {
             return pages.map(Page::progress).average().roundToInt()
         }
 
-    fun setStatusSubject(subject: PublishSubject<Download>?) {
+    enum class State(val value: Int) {
-        statusSubject = subject
+        CHECKED(-1),
-    }
+        NOT_DOWNLOADED(0),
-
+        QUEUE(1),
-    fun setStatusCallback(f: ((Download) -> Unit)?) {
+        DOWNLOADING(2),
-        statusCallback = f
+        DOWNLOADED(3),
-    }
+        ERROR(4),
-
-    enum class State {
-        CHECKED,
-        NOT_DOWNLOADED,
-        QUEUE,
-        DOWNLOADING,
-        DOWNLOADED,
-        ERROR,
         ;
 
         companion object {

app/src/main/java/eu/kanade/tachiyomi/data/download/model/DownloadQueue.kt

@@ -1,131 +1,84 @@
 package eu.kanade.tachiyomi.data.download.model
 
-import com.jakewharton.rxrelay.PublishRelay
+import androidx.annotation.CallSuper
-import eu.kanade.tachiyomi.data.database.models.Chapter
+import eu.kanade.tachiyomi.source.model.Page
-import eu.kanade.tachiyomi.data.download.DownloadStore
+import eu.kanade.tachiyomi.util.system.launchUI
-import eu.kanade.tachiyomi.domain.manga.models.Manga
+import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.MainScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.collectLatest
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.flow.combine
-import rx.subjects.PublishSubject
+import kotlinx.coroutines.flow.debounce
-import java.util.concurrent.*
+import kotlinx.coroutines.flow.distinctUntilChanged
 
-class DownloadQueue(
+sealed class DownloadQueue {
-    private val store: DownloadStore,
+    interface Listener {
-    private val queue: MutableList<Download> = CopyOnWriteArrayList<Download>(),
+        val progressJobs: MutableMap<Download, Job>
-) :
-    List<Download> by queue {
 
-    private val statusSubject = PublishSubject.create<Download>()
+        // Override with presenterScope or viewScope
+        val queueListenerScope: CoroutineScope
 
-    private val updatedRelay = PublishRelay.create<Unit>()
+        fun onPageProgressUpdate(download: Download) {
-
+            onProgressUpdate(download)
-    private val downloadListeners: MutableList<DownloadListener> = CopyOnWriteArrayList<DownloadListener>()
-
-    private var scope = MainScope()
-
-    fun addAll(downloads: List<Download>) {
-        downloads.forEach { download ->
-            download.setStatusSubject(statusSubject)
-            download.setStatusCallback(::setPagesFor)
-            download.status = Download.State.QUEUE
-        }
-        queue.addAll(downloads)
-        store.addAll(downloads)
-        updatedRelay.call(Unit)
         }
+        fun onProgressUpdate(download: Download)
+        fun onQueueUpdate(download: Download)
 
-    fun remove(download: Download) {
+        // Subscribe on presenter/controller creation on UI thread
-        val removed = queue.remove(download)
+        @CallSuper
-        store.remove(download)
+        fun onStatusChange(download: Download) {
-        download.setStatusSubject(null)
+            when (download.status) {
-        download.setStatusCallback(null)
+                Download.State.DOWNLOADING -> {
-        if (download.status == Download.State.DOWNLOADING || download.status == Download.State.QUEUE) {
+                    launchProgressJob(download)
-            download.status = Download.State.NOT_DOWNLOADED
+                    // Initial update of the downloaded pages
+                    onQueueUpdate(download)
+                }
+                Download.State.DOWNLOADED -> {
+                    cancelProgressJob(download)
+
+                    onProgressUpdate(download)
+                    onQueueUpdate(download)
+                }
+                Download.State.ERROR -> cancelProgressJob(download)
+                else -> {
+                    /* unused */
                 }
-        callListeners(download)
-        if (removed) {
-            updatedRelay.call(Unit)
             }
         }
 
-    fun updateListeners() {
+        /**
-        val listeners = downloadListeners.toList()
+         * Observe the progress of a download and notify the view.
-        listeners.forEach { it.updateDownloads() }
+         *
+         * @param download the download to observe its progress.
+         */
+        private fun launchProgressJob(download: Download) {
+            val job = queueListenerScope.launchUI {
+                while (download.pages == null) {
+                    delay(50)
                 }
 
-    fun remove(chapter: Chapter) {
+                val progressFlows = download.pages!!.map(Page::progressFlow)
-        find { it.chapter.id == chapter.id }?.let { remove(it) }
+                combine(progressFlows, Array<Int>::sum)
-    }
+                    .distinctUntilChanged()
-
+                    .debounce(50)
-    fun remove(chapters: List<Chapter>) {
+                    .collectLatest {
-        for (chapter in chapters) { remove(chapter) }
+                        onPageProgressUpdate(download)
-    }
-
-    fun remove(manga: Manga) {
-        filter { it.manga.id == manga.id }.forEach { remove(it) }
-    }
-
-    fun clear() {
-        queue.forEach { download ->
-            download.setStatusSubject(null)
-            download.setStatusCallback(null)
-            if (download.status == Download.State.DOWNLOADING || download.status == Download.State.QUEUE) {
-                download.status = Download.State.NOT_DOWNLOADED
-            }
-            callListeners(download)
-        }
-        queue.clear()
-        store.clear()
-        updatedRelay.call(Unit)
-    }
-
-    private fun setPagesFor(download: Download) {
-        if (download.status == Download.State.DOWNLOADING) {
-            if (download.pages != null) {
-                for (page in download.pages!!)
-                    scope.launch {
-                        page.statusFlow.collectLatest {
-                            callListeners(download)
-                        }
-                    }
-            }
-            callListeners(download)
-        } else if (download.status == Download.State.DOWNLOADED || download.status == Download.State.ERROR) {
-//            setPagesSubject(download.pages, null)
-            if (download.status == Download.State.ERROR) {
-                callListeners(download)
-            }
-        } else {
-            callListeners(download)
                     }
             }
 
-    private fun callListeners(download: Download) {
+            // Avoid leaking jobs
-        val iterator = downloadListeners.iterator()
+            progressJobs.remove(download)?.cancel()
-        while (iterator.hasNext()) {
+
-            iterator.next().updateDownload(download)
+            progressJobs[download] = job
-        }
         }
 
-//    private fun setPagesSubject(pages: List<Page>?, subject: PublishSubject<Int>?) {
+        /**
-//        if (pages != null) {
+         * Unsubscribes the given download from the progress subscriptions.
-//            for (page in pages) {
+         *
-//                page.setStatusSubject(subject)
+         * @param download the download to unsubscribe.
-//            }
+         */
-//        }
+        private fun cancelProgressJob(download: Download) {
-//    }
+            progressJobs.remove(download)?.cancel()
-
+        }
-    fun addListener(listener: DownloadListener) {
-        downloadListeners.add(listener)
-    }
-
-    fun removeListener(listener: DownloadListener) {
-        downloadListeners.remove(listener)
-    }
-
-    interface DownloadListener {
-        fun updateDownload(download: Download)
-        fun updateDownloads()
     }
 }

app/src/main/java/eu/kanade/tachiyomi/data/notification/NotificationReceiver.kt

@@ -64,7 +64,7 @@ class NotificationReceiver : BroadcastReceiver() {
                 downloadManager.pauseDownloads()
             }
             // Clear the download queue
-            ACTION_CLEAR_DOWNLOADS -> downloadManager.clearQueue(true)
+            ACTION_CLEAR_DOWNLOADS -> downloadManager.clearQueue()
             // Delete image from path and dismiss notification
             ACTION_DELETE_IMAGE -> deleteImage(
                 context,

app/src/main/java/eu/kanade/tachiyomi/ui/base/presenter/BaseCoroutinePresenter.kt

@@ -1,10 +1,11 @@
 package eu.kanade.tachiyomi.ui.base.presenter
 
+import androidx.annotation.CallSuper
+import java.lang.ref.WeakReference
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
-import java.lang.ref.WeakReference
 
 open class BaseCoroutinePresenter<T> {
     var presenterScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
@@ -24,6 +25,7 @@ open class BaseCoroutinePresenter<T> {
     open fun onCreate() {
     }
 
+    @CallSuper
     open fun onDestroy() {
         presenterScope.cancel()
         weakView = null

app/src/main/java/eu/kanade/tachiyomi/ui/download/DownloadBottomPresenter.kt

@@ -4,7 +4,9 @@ import eu.kanade.tachiyomi.data.download.DownloadManager
 import eu.kanade.tachiyomi.data.download.model.Download
 import eu.kanade.tachiyomi.data.download.model.DownloadQueue
 import eu.kanade.tachiyomi.ui.base.presenter.BaseCoroutinePresenter
+import eu.kanade.tachiyomi.util.system.launchUI
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import uy.kohesive.injekt.injectLazy
@@ -12,7 +14,8 @@ import uy.kohesive.injekt.injectLazy
 /**
  * Presenter of [DownloadBottomSheet].
  */
-class DownloadBottomPresenter : BaseCoroutinePresenter<DownloadBottomSheet>() {
+class DownloadBottomPresenter : BaseCoroutinePresenter<DownloadBottomSheet>(),
+    DownloadQueue.Listener {
 
     /**
      * Download manager.
@@ -20,15 +23,27 @@ class DownloadBottomPresenter : BaseCoroutinePresenter<DownloadBottomSheet>() {
     val downloadManager: DownloadManager by injectLazy()
     var items = listOf<DownloadHeaderItem>()
 
+    override val progressJobs = mutableMapOf<Download, Job>()
+    override val queueListenerScope get() = presenterScope
+
     /**
      * Property to get the queue from the download manager.
      */
-    val downloadQueue: DownloadQueue
+    val downloadQueueState
-        get() = downloadManager.queue
+        get() = downloadManager.queueState
+
+    override fun onCreate() {
+        presenterScope.launchUI {
+            downloadManager.statusFlow().collect(::onStatusChange)
+        }
+        presenterScope.launchUI {
+            downloadManager.progressFlow().collect(::onPageProgressUpdate)
+        }
+    }
 
     fun getItems() {
         presenterScope.launch {
-            val items = downloadQueue
+            val items = downloadQueueState.value
                 .groupBy { it.source }
                 .map { entry ->
                     DownloadHeaderItem(entry.key.id, entry.key.name, entry.value.size).apply {
@@ -85,4 +100,22 @@ class DownloadBottomPresenter : BaseCoroutinePresenter<DownloadBottomSheet>() {
     fun cancelDownloads(downloads: List<Download>) {
         downloadManager.deletePendingDownloads(*downloads.toTypedArray())
     }
+
+    override fun onStatusChange(download: Download) {
+        super.onStatusChange(download)
+        view?.update(downloadManager.isRunning)
+    }
+
+    override fun onQueueUpdate(download: Download) {
+        view?.onUpdateDownloadedPages(download)
+    }
+
+    override fun onProgressUpdate(download: Download) {
+        view?.onUpdateProgress(download)
+    }
+
+    override fun onPageProgressUpdate(download: Download) {
+        super.onPageProgressUpdate(download)
+        view?.onUpdateDownloadedPages(download)
+    }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/download/DownloadBottomSheet.kt

@@ -117,14 +117,14 @@ class DownloadBottomSheet @JvmOverloads constructor(
     fun update(isRunning: Boolean) {
         presenter.getItems()
         onQueueStatusChange(isRunning)
-        if (binding.downloadFab.isInvisible != presenter.downloadQueue.isEmpty()) {
+        if (binding.downloadFab.isInvisible != presenter.downloadQueueState.value.isEmpty()) {
-            binding.downloadFab.isInvisible = presenter.downloadQueue.isEmpty()
+            binding.downloadFab.isInvisible = presenter.downloadQueueState.value.isEmpty()
         }
         prepareMenu()
     }
 
     private fun updateDLTitle() {
-        val extCount = presenter.downloadQueue.firstOrNull()
+        val extCount = presenter.downloadQueueState.value.firstOrNull()
         binding.titleText.text = if (extCount != null) {
             context.getString(
                 MR.strings.downloading_,
@@ -143,8 +143,8 @@ class DownloadBottomSheet @JvmOverloads constructor(
     private fun onQueueStatusChange(running: Boolean) {
         val oldRunning = isRunning
         isRunning = running
-        if (binding.downloadFab.isInvisible != presenter.downloadQueue.isEmpty()) {
+        if (binding.downloadFab.isInvisible != presenter.downloadQueueState.value.isEmpty()) {
-            binding.downloadFab.isInvisible = presenter.downloadQueue.isEmpty()
+            binding.downloadFab.isInvisible = presenter.downloadQueueState.value.isEmpty()
         }
         updateFab()
         if (oldRunning != running) {
@@ -210,7 +210,7 @@ class DownloadBottomSheet @JvmOverloads constructor(
     private fun setInformationView() {
         updateDLTitle()
         setBottomSheet()
-        if (presenter.downloadQueue.isEmpty()) {
+        if (presenter.downloadQueueState.value.isEmpty()) {
             binding.emptyView.show(
                 R.drawable.ic_download_off_24dp,
                 MR.strings.nothing_is_downloading,
@@ -224,10 +224,10 @@ class DownloadBottomSheet @JvmOverloads constructor(
         val menu = binding.sheetToolbar.menu
         updateFab()
         // Set clear button visibility.
-        menu.findItem(R.id.clear_queue)?.isVisible = !presenter.downloadQueue.isEmpty()
+        menu.findItem(R.id.clear_queue)?.isVisible = presenter.downloadQueueState.value.isNotEmpty()
 
         // Set reorder button visibility.
-        menu.findItem(R.id.reorder)?.isVisible = !presenter.downloadQueue.isEmpty()
+        menu.findItem(R.id.reorder)?.isVisible = presenter.downloadQueueState.value.isNotEmpty()
     }
 
     private fun updateFab() {
@@ -274,7 +274,7 @@ class DownloadBottomSheet @JvmOverloads constructor(
     }
 
     private fun setBottomSheet() {
-        val hasQueue = presenter.downloadQueue.isNotEmpty()
+        val hasQueue = presenter.downloadQueueState.value.isNotEmpty()
         if (hasQueue) {
             sheetBehavior?.skipCollapsed = !hasQueue
             if (sheetBehavior.isHidden()) sheetBehavior?.collapse()
@@ -320,7 +320,6 @@ class DownloadBottomSheet @JvmOverloads constructor(
             }
         }
         presenter.reorder(downloads)
-        controller?.updateChapterDownload(download, false)
     }
 
     /**

app/src/main/java/eu/kanade/tachiyomi/ui/download/DownloadHolder.kt

@@ -68,7 +68,7 @@ class DownloadHolder(private val view: View, val adapter: DownloadAdapter) :
         if (binding.downloadProgress.max == 1) {
             binding.downloadProgress.max = pages.size * 100
         }
-        binding.downloadProgress.progress = download.pageProgress
+        binding.downloadProgress.setProgressCompat(download.pageProgress, true)
     }
 
     /**

app/src/main/java/eu/kanade/tachiyomi/ui/extension/ExtensionBottomPresenter.kt

@@ -2,8 +2,6 @@ package eu.kanade.tachiyomi.ui.extension
 
 import android.content.pm.PackageInstaller
 import eu.kanade.tachiyomi.data.download.DownloadManager
-import eu.kanade.tachiyomi.data.download.model.Download
-import eu.kanade.tachiyomi.data.download.model.DownloadQueue
 import eu.kanade.tachiyomi.extension.ExtensionInstallerJob
 import eu.kanade.tachiyomi.extension.ExtensionManager
 import eu.kanade.tachiyomi.extension.model.Extension
@@ -12,7 +10,6 @@ import eu.kanade.tachiyomi.extension.model.InstalledExtensionsOrder
 import eu.kanade.tachiyomi.extension.util.ExtensionLoader
 import eu.kanade.tachiyomi.ui.migration.BaseMigrationPresenter
 import eu.kanade.tachiyomi.util.system.LocaleHelper
-import eu.kanade.tachiyomi.util.system.launchUI
 import eu.kanade.tachiyomi.util.system.withUIContext
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
@@ -31,7 +28,7 @@ typealias ExtensionIntallInfo = Pair<InstallStep, PackageInstaller.SessionInfo?>
 /**
  * Presenter of [ExtensionBottomSheet].
  */
-class ExtensionBottomPresenter : BaseMigrationPresenter<ExtensionBottomSheet>(), DownloadQueue.DownloadListener {
+class ExtensionBottomPresenter : BaseMigrationPresenter<ExtensionBottomSheet>() {
 
     private var extensions = emptyList<ExtensionItem>()
 
@@ -43,7 +40,7 @@ class ExtensionBottomPresenter : BaseMigrationPresenter<ExtensionBottomSheet>(),
 
     override fun onCreate() {
         super.onCreate()
-        downloadManager.addListener(this)
+
         presenterScope.launch {
             val extensionJob = async {
                 extensionManager.findAvailableExtensions()
@@ -289,11 +286,4 @@ class ExtensionBottomPresenter : BaseMigrationPresenter<ExtensionBottomSheet>(),
             extensionManager.trust(pkgName, versionCode, signatureHash)
         }
     }
-
-    override fun updateDownload(download: Download) = updateDownloads()
-    override fun updateDownloads() {
-        presenterScope.launchUI {
-            view?.updateDownloadStatus(!downloadManager.isPaused())
-        }
-    }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/extension/ExtensionBottomSheet.kt

@@ -521,8 +521,4 @@ class ExtensionBottomSheet @JvmOverloads constructor(context: Context, attrs: At
             return if (index == -1) POSITION_NONE else index
         }
     }
-
-    fun updateDownloadStatus(isRunning: Boolean) {
-        (controller.activity as? MainActivity)?.downloadStatusChanged(isRunning)
-    }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/library/LibraryController.kt

@@ -61,7 +61,6 @@ import eu.kanade.tachiyomi.R
 import eu.kanade.tachiyomi.core.preference.Preference
 import eu.kanade.tachiyomi.data.database.models.Category
 import eu.kanade.tachiyomi.data.database.models.LibraryManga
-import eu.kanade.tachiyomi.data.download.DownloadJob
 import eu.kanade.tachiyomi.data.library.LibraryUpdateJob
 import eu.kanade.tachiyomi.data.notification.NotificationReceiver
 import eu.kanade.tachiyomi.data.notification.Notifications
@@ -1059,7 +1058,6 @@ open class LibraryController(
                 presenter.getLibrary()
                 isPoppingIn = true
             }
-            DownloadJob.callListeners()
             binding.recyclerCover.isClickable = false
             binding.recyclerCover.isFocusable = false
             singleCategory = presenter.categories.size <= 1
@@ -2199,8 +2197,4 @@ open class LibraryController(
             }
         }
     }
-
-    fun updateDownloadStatus(isRunning: Boolean) {
-        (activity as? MainActivity)?.downloadStatusChanged(isRunning)
-    }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/library/LibraryPresenter.kt

@@ -11,8 +11,6 @@ import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.data.database.models.removeCover
 import eu.kanade.tachiyomi.data.database.models.seriesType
 import eu.kanade.tachiyomi.data.download.DownloadManager
-import eu.kanade.tachiyomi.data.download.model.Download
-import eu.kanade.tachiyomi.data.download.model.DownloadQueue
 import eu.kanade.tachiyomi.data.preference.DelayedLibrarySuggestionsJob
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.data.track.TrackManager
@@ -44,7 +42,6 @@ import eu.kanade.tachiyomi.util.manga.MangaCoverMetadata
 import eu.kanade.tachiyomi.util.mapStatus
 import eu.kanade.tachiyomi.util.system.launchIO
 import eu.kanade.tachiyomi.util.system.launchNonCancellableIO
-import eu.kanade.tachiyomi.util.system.launchUI
 import eu.kanade.tachiyomi.util.system.withIOContext
 import eu.kanade.tachiyomi.util.system.withUIContext
 import java.util.*
@@ -95,7 +92,7 @@ class LibraryPresenter(
     private val downloadManager: DownloadManager = Injekt.get(),
     private val chapterFilter: ChapterFilter = Injekt.get(),
     private val trackManager: TrackManager = Injekt.get(),
-) : BaseCoroutinePresenter<LibraryController>(), DownloadQueue.DownloadListener {
+) : BaseCoroutinePresenter<LibraryController>() {
     private val getCategories: GetCategories by injectLazy()
     private val setMangaCategories: SetMangaCategories by injectLazy()
     private val updateCategories: UpdateCategories by injectLazy()
@@ -189,7 +186,7 @@ class LibraryPresenter(
 
     override fun onCreate() {
         super.onCreate()
-        downloadManager.addListener(this)
+
         if (!controllerIsSubClass) {
             lastLibraryItems?.let { libraryItems = it }
             lastCategories?.let { categories = it }
@@ -1640,13 +1637,6 @@ class LibraryPresenter(
         }
     }
 
-    override fun updateDownload(download: Download) = updateDownloads()
-    override fun updateDownloads() {
-        presenterScope.launchUI {
-            view?.updateDownloadStatus(!downloadManager.isPaused())
-        }
-    }
-
     data class ItemPreferences(
         val filterDownloaded: Int,
         val filterUnread: Int,

app/src/main/java/eu/kanade/tachiyomi/ui/main/MainActivity.kt

@@ -62,12 +62,12 @@ import com.bluelinelabs.conductor.ControllerChangeHandler
 import com.bluelinelabs.conductor.Router
 import com.getkeepsafe.taptargetview.TapTarget
 import com.getkeepsafe.taptargetview.TapTargetView
+import com.google.android.material.badge.BadgeDrawable
 import com.google.android.material.navigation.NavigationBarView
 import com.google.android.material.snackbar.Snackbar
 import com.google.android.material.transition.platform.MaterialContainerTransformSharedElementCallback
 import eu.kanade.tachiyomi.BuildConfig
 import eu.kanade.tachiyomi.R
-import eu.kanade.tachiyomi.data.download.DownloadJob
 import eu.kanade.tachiyomi.data.download.DownloadManager
 import eu.kanade.tachiyomi.data.library.LibraryUpdateJob
 import eu.kanade.tachiyomi.data.notification.NotificationReceiver
@@ -458,7 +458,7 @@ open class MainActivity : BaseActivity<MainActivityBinding>() {
             }
         }
 
-        DownloadJob.downloadFlow.onEach(::downloadStatusChanged).launchIn(lifecycleScope)
+        downloadManager.isDownloaderRunning.onEach(::downloadStatusChanged).launchIn(lifecycleScope)
         lifecycleScope
         WindowCompat.setDecorFitsSystemWindows(window, false)
         setSupportActionBar(binding.toolbar)
@@ -947,7 +947,6 @@ open class MainActivity : BaseActivity<MainActivityBinding>() {
             extensionManager.getExtensionUpdates(false)
         }
         setExtensionsBadge()
-        DownloadJob.callListeners(downloadManager = downloadManager)
         showDLQueueTutorial()
         reEnableBackPressedCallBack()
     }
@@ -1504,12 +1503,16 @@ open class MainActivity : BaseActivity<MainActivityBinding>() {
         }
     }
 
+    fun BadgeDrawable.updateQueueSize(queueSize: Int) {
+        number = queueSize
+    }
+
     fun downloadStatusChanged(downloading: Boolean) {
         lifecycleScope.launchUI {
             val hasQueue = downloading || downloadManager.hasQueue()
             if (hasQueue) {
                 val badge = nav.getOrCreateBadge(R.id.nav_recents)
-                badge.number = downloadManager.queue.size
+                badge.updateQueueSize(downloadManager.queueState.value.size)
                 if (downloading) badge.backgroundColor = -870219 else badge.backgroundColor = Color.GRAY
                 showDLQueueTutorial()
             } else {

app/src/main/java/eu/kanade/tachiyomi/ui/manga/MangaDetailsController.kt

@@ -792,7 +792,6 @@ class MangaDetailsController :
         binding.swipeRefresh.isRefreshing = enabled
     }
 
-    //region Recycler methods
     fun updateChapterDownload(download: Download) {
         getHolder(download.chapter)?.notifyStatus(
             download.status,
@@ -1802,7 +1801,7 @@ class MangaDetailsController :
     override fun onDestroyActionMode(mode: ActionMode?) {
         actionMode = null
         setStatusBarAndToolbar()
-        if (startingRangeChapterPos != null && rangeMode == RangeMode.Download) {
+        if (startingRangeChapterPos != null && rangeMode in setOf(RangeMode.Download, RangeMode.RemoveDownload)) {
             val item = adapter?.getItem(startingRangeChapterPos!!) as? ChapterItem
             (binding.recycler.findViewHolderForAdapterPosition(startingRangeChapterPos!!) as? ChapterHolder)?.notifyStatus(
                 item?.status ?: Download.State.NOT_DOWNLOADED,

app/src/main/java/eu/kanade/tachiyomi/ui/manga/MangaDetailsPresenter.kt

@@ -60,6 +60,7 @@ import eu.kanade.tachiyomi.util.manga.MangaUtil
 import eu.kanade.tachiyomi.util.shouldDownloadNewChapters
 import eu.kanade.tachiyomi.util.storage.DiskUtil
 import eu.kanade.tachiyomi.util.system.ImageUtil
+import eu.kanade.tachiyomi.util.system.e
 import eu.kanade.tachiyomi.util.system.launchIO
 import eu.kanade.tachiyomi.util.system.launchNonCancellableIO
 import eu.kanade.tachiyomi.util.system.launchNow
@@ -72,8 +73,11 @@ import java.io.FileOutputStream
 import java.io.OutputStream
 import java.util.Locale
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
@@ -108,7 +112,8 @@ class MangaDetailsPresenter(
     private val downloadManager: DownloadManager = Injekt.get(),
     private val chapterFilter: ChapterFilter = Injekt.get(),
     private val storageManager: StorageManager = Injekt.get(),
-) : BaseCoroutinePresenter<MangaDetailsController>(), DownloadQueue.DownloadListener {
+) : BaseCoroutinePresenter<MangaDetailsController>(),
+    DownloadQueue.Listener {
     private val getAvailableScanlators: GetAvailableScanlators by injectLazy()
     private val getCategories: GetCategories by injectLazy()
     private val getChapter: GetChapter by injectLazy()
@@ -174,6 +179,9 @@ class MangaDetailsPresenter(
 
     var allChapterScanlators: Set<String> = emptySet()
 
+    override val progressJobs: MutableMap<Download, Job> = mutableMapOf()
+    override val queueListenerScope get() = presenterScope
+
     override fun onCreate() {
         val controller = view ?: return
 
@@ -181,10 +189,24 @@ class MangaDetailsPresenter(
         if (!::manga.isInitialized) runBlocking { refreshMangaFromDb() }
         syncData()
 
-        downloadManager.addListener(this)
+        presenterScope.launchUI {
+            downloadManager.statusFlow()
+                .filter { it.manga.id == mangaId }
+                .catch { error -> Logger.e(error) }
+                .collect(::onStatusChange)
+        }
+        presenterScope.launchUI {
+            downloadManager.progressFlow()
+                .filter { it.manga.id == mangaId }
+                .catch { error -> Logger.e(error) }
+                .collect(::onQueueUpdate)
+        }
+        presenterScope.launchIO {
+            downloadManager.queueState.collectLatest(::onQueueUpdate)
+        }
 
         runBlocking {
-            tracks = getTrack.awaitAllByMangaId(manga.id!!)
+            tracks = getTrack.awaitAllByMangaId(mangaId)
         }
     }
 
@@ -219,11 +241,6 @@ class MangaDetailsPresenter(
         refreshTracking(false)
     }
 
-    override fun onDestroy() {
-        super.onDestroy()
-        downloadManager.removeListener(this)
-    }
-
     fun fetchChapters(andTracking: Boolean = true) {
         presenterScope.launch {
             getChapters()
@@ -252,12 +269,12 @@ class MangaDetailsPresenter(
         return chapters
     }
 
-    private suspend fun getChapters() {
+    private suspend fun getChapters(queue: List<Download> = downloadManager.queueState.value) {
         val chapters = getChapter.awaitAll(mangaId, isScanlatorFiltered()).map { it.toModel() }
         allChapters = if (!isScanlatorFiltered()) chapters else getChapter.awaitAll(mangaId, false).map { it.toModel() }
 
         // Find downloaded chapters
-        setDownloadedChapters(chapters)
+        setDownloadedChapters(chapters, queue)
         allChapterScanlators = allChapters.mapNotNull { it.chapter.scanlator }.toSet()
 
         this.chapters = applyChapterFilters(chapters)
@@ -274,33 +291,17 @@ class MangaDetailsPresenter(
      *
      * @param chapters the list of chapter from the database.
      */
-    private fun setDownloadedChapters(chapters: List<ChapterItem>) {
+    private fun setDownloadedChapters(chapters: List<ChapterItem>, queue: List<Download>) {
         for (chapter in chapters) {
             if (downloadManager.isChapterDownloaded(chapter, manga)) {
                 chapter.status = Download.State.DOWNLOADED
-            } else if (downloadManager.hasQueue()) {
+            } else if (queue.isNotEmpty()) {
-                chapter.status = downloadManager.queue.find { it.chapter.id == chapter.id }
+                chapter.status = queue.find { it.chapter.id == chapter.id }
                     ?.status ?: Download.State.default
             }
         }
     }
 
-    override fun updateDownload(download: Download) {
-        chapters.find { it.id == download.chapter.id }?.download = download
-        presenterScope.launchUI {
-            view?.updateChapterDownload(download)
-        }
-    }
-
-    override fun updateDownloads() {
-        presenterScope.launch(Dispatchers.Default) {
-            getChapters()
-            withContext(Dispatchers.Main) {
-                view?.updateChapters(chapters)
-            }
-        }
-    }
-
     /**
      * Converts a chapter from the database to an extended model, allowing to store new fields.
      */
@@ -310,7 +311,7 @@ class MangaDetailsPresenter(
         model.isLocked = isLockedFromSearch
 
         // Find an active download for this chapter.
-        val download = downloadManager.queue.find { it.chapter.id == id }
+        val download = downloadManager.queueState.value.find { it.chapter.id == id }
 
         if (download != null) {
             // If there's an active download, assign it.
@@ -387,14 +388,15 @@ class MangaDetailsPresenter(
      * @param chapter the chapter to delete.
      */
     fun deleteChapter(chapter: ChapterItem) {
-        downloadManager.deleteChapters(listOf(chapter), manga, source, true)
         this.chapters.find { it.id == chapter.id }?.apply {
             if (chapter.chapter.bookmark && !preferences.removeBookmarkedChapters().get()) return@apply
-            status = Download.State.QUEUE
+            status = Download.State.NOT_DOWNLOADED
             download = null
         }
 
         view?.updateChapters(this.chapters)
+
+        downloadManager.deleteChapters(listOf(chapter), manga, source, true)
     }
 
     /**
@@ -402,22 +404,21 @@ class MangaDetailsPresenter(
      * @param chapters the list of chapters to delete.
      */
     fun deleteChapters(chapters: List<ChapterItem>, update: Boolean = true, isEverything: Boolean = false) {
-        launchIO {
-            if (isEverything) {
-                downloadManager.deleteManga(manga, source)
-            } else {
-                downloadManager.deleteChapters(chapters, manga, source)
-            }
-        }
         chapters.forEach { chapter ->
             this.chapters.find { it.id == chapter.id }?.apply {
                 if (chapter.chapter.bookmark && !preferences.removeBookmarkedChapters().get() && !isEverything) return@apply
-                status = Download.State.QUEUE
+                status = Download.State.NOT_DOWNLOADED
                 download = null
             }
         }
 
         if (update) view?.updateChapters(this.chapters)
+
+        if (isEverything) {
+            downloadManager.deleteManga(manga, source)
+        } else {
+            downloadManager.deleteChapters(chapters, manga, source)
+        }
     }
 
     suspend fun refreshMangaFromDb(): Manga {
@@ -1150,6 +1151,32 @@ class MangaDetailsPresenter(
         return if (date <= 0L) null else date
     }
 
+    override fun onStatusChange(download: Download) {
+        super.onStatusChange(download)
+        chapters.find { it.id == download.chapter.id }?.status = download.status
+        onPageProgressUpdate(download)
+    }
+
+    private suspend fun onQueueUpdate(queue: List<Download>) = withIOContext {
+        getChapters(queue)
+        withUIContext {
+            view?.updateChapters(chapters)
+        }
+    }
+
+    override fun onQueueUpdate(download: Download) {
+        // already handled by onStatusChange
+    }
+
+    override fun onProgressUpdate(download: Download) {
+        // already handled by onStatusChange
+    }
+
+    override fun onPageProgressUpdate(download: Download) {
+        chapters.find { it.id == download.chapter.id }?.download = download
+        view?.updateChapterDownload(download)
+    }
+
     companion object {
         const val MULTIPLE_VOLUMES = 1
         const val TENS_OF_CHAPTERS = 2

app/src/main/java/eu/kanade/tachiyomi/ui/recents/RecentsController.kt

@@ -629,13 +629,8 @@ class RecentsController(bundle: Bundle? = null) :
         }
     }
 
-    fun updateChapterDownload(download: Download, updateDLSheet: Boolean = true) {
+    fun updateChapterDownload(download: Download) {
-        if (view == null) return
+        if (view == null || !this::adapter.isInitialized) return
-        if (updateDLSheet) {
-            binding.downloadBottomSheet.dlBottomSheet.update(!presenter.downloadManager.isPaused())
-            binding.downloadBottomSheet.dlBottomSheet.onUpdateProgress(download)
-            binding.downloadBottomSheet.dlBottomSheet.onUpdateDownloadedPages(download)
-        }
         val id = download.chapter.id ?: return
         val item = adapter.getItemByChapterId(id) ?: return
         val holder = binding.recycler.findViewHolderForItemId(item.id!!) as? RecentMangaHolder ?: return

app/src/main/java/eu/kanade/tachiyomi/ui/recents/RecentsPresenter.kt

@@ -5,7 +5,6 @@ import eu.kanade.tachiyomi.data.database.models.ChapterHistory
 import eu.kanade.tachiyomi.data.database.models.History
 import eu.kanade.tachiyomi.data.database.models.HistoryImpl
 import eu.kanade.tachiyomi.data.database.models.MangaChapterHistory
-import eu.kanade.tachiyomi.data.download.DownloadJob
 import eu.kanade.tachiyomi.data.download.DownloadManager
 import eu.kanade.tachiyomi.data.download.model.Download
 import eu.kanade.tachiyomi.data.download.model.DownloadQueue
@@ -31,6 +30,7 @@ import kotlin.math.abs
 import kotlin.math.roundToInt
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.drop
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
@@ -55,7 +55,8 @@ class RecentsPresenter(
     val preferences: PreferencesHelper = Injekt.get(),
     val downloadManager: DownloadManager = Injekt.get(),
     private val chapterFilter: ChapterFilter = Injekt.get(),
-) : BaseCoroutinePresenter<RecentsController>(), DownloadQueue.DownloadListener {
+) : BaseCoroutinePresenter<RecentsController>(),
+    DownloadQueue.Listener {
     private val handler: DatabaseHandler by injectLazy()
 
     private val getChapter: GetChapter by injectLazy()
@@ -99,10 +100,27 @@ class RecentsPresenter(
     private val isOnFirstPage: Boolean
         get() = pageOffset == 0
 
+    override val progressJobs = mutableMapOf<Download, Job>()
+    override val queueListenerScope get() = presenterScope
+
     override fun onCreate() {
         super.onCreate()
-        downloadManager.addListener(this)
+        presenterScope.launchUI {
-        DownloadJob.downloadFlow.onEach(::downloadStatusChanged).launchIn(presenterScope)
+            downloadManager.statusFlow().collect(::onStatusChange)
+        }
+        presenterScope.launchUI {
+            downloadManager.progressFlow().collect(::onProgressUpdate)
+        }
+        presenterScope.launchIO {
+            downloadManager.queueState.collectLatest {
+                setDownloadedChapters(recentItems, it)
+                withUIContext {
+                    view?.showLists(recentItems, true)
+                    view?.updateDownloadStatus(!downloadManager.isPaused())
+                }
+            }
+        }
+        downloadManager.isDownloaderRunning.onEach(::downloadStatusChanged).launchIn(presenterScope)
         LibraryUpdateJob.updateFlow.onEach(::onUpdateManga).launchIn(presenterScope)
         if (lastRecents != null) {
             if (recentItems.isEmpty()) {
@@ -482,7 +500,6 @@ class RecentsPresenter(
 
     override fun onDestroy() {
         super.onDestroy()
-        downloadManager.removeListener(this)
         lastRecents = recentItems
     }
 
@@ -500,12 +517,12 @@ class RecentsPresenter(
      *
      * @param chapters the list of chapter from the database.
      */
-    private fun setDownloadedChapters(chapters: List<RecentMangaItem>) {
+    private fun setDownloadedChapters(chapters: List<RecentMangaItem>, queue: List<Download> = downloadManager.queueState.value) {
         for (item in chapters.filter { it.chapter.id != null }) {
             if (downloadManager.isChapterDownloaded(item.chapter, item.mch.manga)) {
                 item.status = Download.State.DOWNLOADED
-            } else if (downloadManager.hasQueue()) {
+            } else if (queue.isNotEmpty()) {
-                item.download = downloadManager.queue.find { it.chapter.id == item.chapter.id }
+                item.download = queue.find { it.chapter.id == item.chapter.id }
                 item.status = item.download?.status ?: Download.State.default
             }
 
@@ -514,8 +531,8 @@ class RecentsPresenter(
                 downloadInfo.chapterId = chapter.id
                 if (downloadManager.isChapterDownloaded(chapter, item.mch.manga)) {
                     downloadInfo.status = Download.State.DOWNLOADED
-                } else if (downloadManager.hasQueue()) {
+                } else if (queue.isNotEmpty()) {
-                    downloadInfo.download = downloadManager.queue.find { it.chapter.id == chapter.id }
+                    downloadInfo.download = queue.find { it.chapter.id == chapter.id }
                     downloadInfo.status = downloadInfo.download?.status ?: Download.State.default
                 }
                 downloadInfo
@@ -523,32 +540,6 @@ class RecentsPresenter(
         }
     }
 
-    override fun updateDownload(download: Download) {
-        recentItems.find {
-            download.chapter.id == it.chapter.id ||
-                download.chapter.id in it.mch.extraChapters.map { ch -> ch.id }
-        }?.apply {
-            if (chapter.id != download.chapter.id) {
-                val downloadInfo = downloadInfo.find { it.chapterId == download.chapter.id }
-                    ?: return@apply
-                downloadInfo.download = download
-            } else {
-                this.download = download
-            }
-        }
-        presenterScope.launchUI { view?.updateChapterDownload(download) }
-    }
-
-    override fun updateDownloads() {
-        presenterScope.launch {
-            setDownloadedChapters(recentItems)
-            withContext(Dispatchers.Main) {
-                view?.showLists(recentItems, true)
-                view?.updateDownloadStatus(!downloadManager.isPaused())
-            }
-        }
-    }
-
     private fun downloadStatusChanged(downloading: Boolean) {
         presenterScope.launchUI {
             view?.updateDownloadStatus(downloading)
@@ -704,6 +695,18 @@ class RecentsPresenter(
         }
     }
 
+    override fun onProgressUpdate(download: Download) {
+        // don't do anything
+    }
+
+    override fun onQueueUpdate(download: Download) {
+        view?.updateChapterDownload(download)
+    }
+
+    override fun onPageProgressUpdate(download: Download) {
+        view?.updateChapterDownload(download)
+    }
+
     enum class GroupType {
         BySeries,
         ByWeek,