Do library checks from up to 5 sources concurrently

app/src/main/java/eu/kanade/tachiyomi/data/library/LibraryUpdateService.kt

@@ -308,36 +308,46 @@ class LibraryUpdateService(
 
         // Emit each manga and update it sequentially.
         return Observable.from(mangaToUpdate)
-            // Notify manga that will update.
-            .doOnNext { showProgressNotification(it, count.andIncrement, mangaToUpdate.size) }
-            // Update the chapters of the manga.
-            .concatMap { manga ->
-                updateManga(manga)
-                    // If there's any error, return empty update and continue.
-                    .onErrorReturn {
-                        failedUpdates.add(manga)
-                        Pair(emptyList(), emptyList())
-                    }
-                    // Filter out mangas without new chapters (or failed).
-                    .filter { pair -> pair.first.isNotEmpty() }
-                    .doOnNext {
-                        if (downloadNew && (
-                            categoriesToDownload.isEmpty() ||
-                                manga.category in categoriesToDownload
-                            )
-                        ) {
-                            downloadChapters(manga, it.first)
-                            hasDownloads = true
+            // Update the chapters of the manga concurrently from 5 different sources
+            .groupBy { it.source }
+            .flatMap(
+                { bySource ->
+                    bySource
+                        // Notify manga that will update.
+                        .doOnNext { showProgressNotification(it, count.andIncrement, mangaToUpdate.size) }
+                        .concatMap { manga ->
+                            updateManga(manga)
+                                // If there's any error, return empty update and continue.
+                                .onErrorReturn {
+                                    failedUpdates.add(manga)
+                                    Pair(emptyList(), emptyList())
+                                }
+                                // Filter out mangas without new chapters (or failed).
+                                .filter { pair -> pair.first.isNotEmpty() }
+                                .doOnNext {
+                                    if (downloadNew && (
+                                        categoriesToDownload.isEmpty() ||
+                                            manga.category in categoriesToDownload
+                                        )
+                                    ) {
+                                        downloadChapters(manga, it.first)
+                                        hasDownloads = true
+                                    }
+                                }
+                                // Convert to the manga that contains new chapters.
+                                .map {
+                                    Pair(
+                                        manga,
+                                        (
+                                            it.first.sortedByDescending { ch -> ch.source_order }
+                                                .toTypedArray()
+                                            )
+                                    )
+                                }
                         }
-                    }
-                    // Convert to the manga that contains new chapters.
-                    .map {
-                        Pair(
-                            manga,
-                            (it.first.sortedByDescending { ch -> ch.source_order }.toTypedArray())
-                        )
-                    }
-            }
+                },
+                5
+            )
             // Add manga with new chapters to the list.
             .doOnNext { manga ->
                 // Add to the list
@@ -397,21 +407,28 @@ class LibraryUpdateService(
 
         // Emit each manga and update it sequentially.
         return Observable.from(mangaToUpdate)
-            // Notify manga that will update.
-            .doOnNext { showProgressNotification(it, count.andIncrement, mangaToUpdate.size) }
-            // Update the details of the manga.
-            .concatMap { manga ->
-                val source = sourceManager.get(manga.source) as? HttpSource
-                    ?: return@concatMap Observable.empty<LibraryManga>()
-
-                source.fetchMangaDetails(manga)
-                    .map { networkManga ->
-                        manga.copyFrom(networkManga)
-                        db.insertManga(manga).executeAsBlocking()
-                        manga
-                    }
-                    .onErrorReturn { manga }
-            }
+            // Update the details of the manga concurrently from 5 different sources
+            .groupBy { it.source }
+            .flatMap(
+                { bySource ->
+                    bySource
+                        // Notify manga that will update.
+                        .doOnNext { showProgressNotification(it, count.andIncrement, mangaToUpdate.size) }
+                        .concatMap { manga ->
+                            val source = sourceManager.get(manga.source) as? HttpSource
+                                ?: return@concatMap Observable.empty<LibraryManga>()
+
+                            source.fetchMangaDetails(manga)
+                                .map { networkManga ->
+                                    manga.copyFrom(networkManga)
+                                    db.insertManga(manga).executeAsBlocking()
+                                    manga
+                                }
+                                .onErrorReturn { manga }
+                        }
+                },
+                5
+            )
             .doOnCompleted {
                 cancelProgressNotification()
             }