Refactor and cleanup

app/src/main/java/eu/kanade/tachiyomi/data/backup/legacy/LegacyBackupManager.kt

@@ -52,13 +52,12 @@ import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.source.LocalSource
 import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.online.all.MergedSource
-import eu.kanade.tachiyomi.util.lang.asObservable
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import exh.MERGED_SOURCE_ID
 import exh.eh.EHentaiThrottleManager
 import exh.merged.sql.models.MergedMangaReference
 import exh.savedsearches.JsonSavedSearch
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.runBlocking
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
@@ -317,13 +316,14 @@ class LegacyBackupManager(context: Context, version: Int = CURRENT_VERSION) : Ab
     override fun restoreChapterFetchObservable(source: Source, manga: Manga, chapters: List<Chapter>, throttleManager: EHentaiThrottleManager): Observable<Pair<List<Chapter>, List<Chapter>>> {
         // SY -->
         return if (source is MergedSource) {
-            val syncedChapters = runBlocking { source.fetchChaptersAndSync(manga, false) }
+            runAsObservable({
-            syncedChapters.onEach { pair ->
+                val syncedChapters = source.fetchChaptersAndSync(manga, false)
-                if (pair.first.isNotEmpty()) {
+                syncedChapters.first.onEach {
-                    chapters.forEach { it.manga_id = manga.id }
+                    it.manga_id = manga.id
-                    updateChapters(chapters)
                 }
-            }.asObservable()
+                updateChapters(syncedChapters.first)
+                syncedChapters
+            })
         } else {
             super.restoreChapterFetchObservable(source, manga, chapters, throttleManager)
         }

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

@@ -30,8 +30,8 @@ import eu.kanade.tachiyomi.ui.library.LibraryGroup
 import eu.kanade.tachiyomi.ui.manga.track.TrackItem
 import eu.kanade.tachiyomi.util.chapter.NoChaptersException
 import eu.kanade.tachiyomi.util.chapter.syncChaptersWithSource
-import eu.kanade.tachiyomi.util.lang.asObservable
 import eu.kanade.tachiyomi.util.lang.awaitSingle
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import eu.kanade.tachiyomi.util.prepUpdateCover
 import eu.kanade.tachiyomi.util.shouldDownloadNewChapters
 import eu.kanade.tachiyomi.util.storage.getUriCompat
@@ -452,7 +452,7 @@ class LibraryUpdateService(
         }
 
         return (
-            /* SY --> */ if (source is MergedSource) runBlocking { source.fetchChaptersAndSync(manga, false).asObservable() }
+            /* SY --> */ if (source is MergedSource) runAsObservable({ source.fetchChaptersAndSync(manga, false) })
             else /* SY <-- */ source.fetchChapterList(manga)
                 .map { syncChaptersWithSource(db, it, manga, source) }
             // SY -->
@@ -544,8 +544,7 @@ class LibraryUpdateService(
     private fun syncFollows(): Observable<LibraryManga> {
         val count = AtomicInteger(0)
         val mangaDex = MdUtil.getEnabledMangaDex(preferences, sourceManager) ?: return Observable.empty()
-        return mangaDex.fetchAllFollows(true)
+        return runAsObservable({ mangaDex.fetchAllFollows(true) })
-            .asObservable()
             .map { listManga ->
                 listManga.filter { (_, metadata) ->
                     metadata.follow_status == FollowStatus.RE_READING.int || metadata.follow_status == FollowStatus.READING.int
@@ -600,7 +599,7 @@ class LibraryUpdateService(
 
                 if (tracker.track?.status == FollowStatus.UNFOLLOWED.int) {
                     tracker.track.status = FollowStatus.READING.int
-                    tracker.service.update(tracker.track)
+                    runAsObservable({ tracker.service.update(tracker.track) })
                 } else Observable.just(null)
             }
             .doOnNext { returnedTracker ->

app/src/main/java/eu/kanade/tachiyomi/data/track/mdlist/MdList.kt

@@ -10,16 +10,15 @@ import eu.kanade.tachiyomi.data.track.TrackManager
 import eu.kanade.tachiyomi.data.track.TrackService
 import eu.kanade.tachiyomi.data.track.model.TrackSearch
 import eu.kanade.tachiyomi.source.model.SManga
-import eu.kanade.tachiyomi.util.lang.asObservable
+import eu.kanade.tachiyomi.util.lang.await
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import exh.md.utils.FollowStatus
 import exh.md.utils.MdUtil
 import exh.metadata.metadata.MangaDexSearchMetadata
 import exh.metadata.metadata.base.getFlatMetadataForManga
 import exh.metadata.metadata.base.insertFlatMetadata
 import exh.util.floor
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.runBlocking
-import rx.Completable
 import rx.Observable
 import uy.kohesive.injekt.injectLazy
 
@@ -49,52 +48,48 @@ class MdList(private val context: Context, id: Int) : TrackService(id) {
 
     override fun displayScore(track: Track) = track.score.toInt().toString()
 
-    override fun add(track: Track): Observable<Track> {
+    override suspend fun add(track: Track): Track {
         return update(track)
     }
 
-    override fun update(track: Track): Observable<Track> {
+    override suspend fun update(track: Track): Track {
         val mdex = mdex ?: throw Exception("Mangadex not enabled")
-        return Observable.defer {
+        val manga = db.getManga(track.tracking_url.substringAfter(".org"), mdex.id).await() ?: throw Exception("Manga doesnt exist")
-            db.getManga(track.tracking_url.substringAfter(".org"), mdex.id)
-                .asRxObservable()
-                .map { manga ->
-                    val mangaMetadata = db.getFlatMetadataForManga(manga.id!!).executeAsBlocking()?.raise(MangaDexSearchMetadata::class) ?: throw Exception("Invalid manga metadata")
-                    val followStatus = FollowStatus.fromInt(track.status)!!
 
-                    // allow follow status to update
+        val mangaMetadata = db.getFlatMetadataForManga(manga.id!!).executeAsBlocking()?.raise(MangaDexSearchMetadata::class) ?: throw Exception("Invalid manga metadata")
-                    if (mangaMetadata.follow_status != followStatus.int) {
+        val followStatus = FollowStatus.fromInt(track.status)!!
-                        runBlocking { mdex.updateFollowStatus(MdUtil.getMangaId(track.tracking_url), followStatus).collect() }
-                        mangaMetadata.follow_status = followStatus.int
-                        db.insertFlatMetadata(mangaMetadata.flatten()).await()
-                    }
 
-                    if (track.score.toInt() > 0) {
+        // allow follow status to update
-                        runBlocking { mdex.updateRating(track).collect() }
+        if (mangaMetadata.follow_status != followStatus.int) {
-                    }
+            mdex.updateFollowStatus(MdUtil.getMangaId(track.tracking_url), followStatus)
-
+            mangaMetadata.follow_status = followStatus.int
-                    // mangadex wont update chapters if manga is not follows this prevents unneeded network call
+            db.insertFlatMetadata(mangaMetadata.flatten()).await()
-
-                    if (followStatus != FollowStatus.UNFOLLOWED) {
-                        if (track.total_chapters != 0 && track.last_chapter_read == track.total_chapters) {
-                            track.status = FollowStatus.COMPLETED.int
-                        }
-
-                        runBlocking { mdex.updateReadingProgress(track).collect() }
-                    } else if (track.last_chapter_read != 0) {
-                        // When followStatus has been changed to unfollowed 0 out read chapters since dex does
-                        track.last_chapter_read = 0
-                    }
-                    track
-                }
         }
+
+        if (track.score.toInt() > 0) {
+            mdex.updateRating(track)
+        }
+
+        // mangadex wont update chapters if manga is not follows this prevents unneeded network call
+
+        if (followStatus != FollowStatus.UNFOLLOWED) {
+            if (track.total_chapters != 0 && track.last_chapter_read == track.total_chapters) {
+                track.status = FollowStatus.COMPLETED.int
+            }
+
+            mdex.updateReadingProgress(track)
+        } else if (track.last_chapter_read != 0) {
+            // When followStatus has been changed to unfollowed 0 out read chapters since dex does
+            track.last_chapter_read = 0
+        }
+        return track
     }
 
     override fun getCompletionStatus(): Int = FollowStatus.COMPLETED.int
 
     override fun bind(track: Track): Observable<Track> {
         val mdex = mdex ?: throw Exception("Mangadex not enabled")
-        return mdex.fetchTrackingInfo(track.tracking_url).asObservable()
+        return runAsObservable({ mdex.fetchTrackingInfo(track.tracking_url) })
             .doOnNext { remoteTrack ->
                 track.copyPersonalFrom(remoteTrack)
                 val manga = db.getManga(track.manga_id).executeAsBlocking()
@@ -103,13 +98,13 @@ class MdList(private val context: Context, id: Int) : TrackService(id) {
                 } else {
                     remoteTrack.total_chapters
                 }
-                update(track)
+                runBlocking { update(track) }
             }
     }
 
     override fun refresh(track: Track): Observable<Track> {
         val mdex = mdex ?: throw Exception("Mangadex not enabled")
-        return mdex.fetchTrackingInfo(track.tracking_url).asObservable()
+        return runAsObservable({ mdex.fetchTrackingInfo(track.tracking_url) })
             .map { remoteTrack ->
                 track.copyPersonalFrom(remoteTrack)
                 val manga = db.getManga(track.manga_id).executeAsBlocking()
@@ -133,5 +128,5 @@ class MdList(private val context: Context, id: Int) : TrackService(id) {
 
     override fun search(query: String): Observable<List<TrackSearch>> = throw Exception("not used")
 
-    override fun login(username: String, password: String): Completable = throw Exception("not used")
+    override suspend fun login(username: String, password: String): Unit = throw Exception("not used")
 }

app/src/main/java/eu/kanade/tachiyomi/extension/ExtensionManager.kt

@@ -154,7 +154,7 @@ class ExtensionManager(
     }
 
     // EXH -->
-    fun <T : Extension> Iterable<T>.filterNotBlacklisted(): List<T> {
+    private fun <T : Extension> Iterable<T>.filterNotBlacklisted(): List<T> {
         val blacklistEnabled = preferences.enableSourceBlacklist().get()
         return filter {
             if (it.isBlacklisted(blacklistEnabled)) {
@@ -164,10 +164,7 @@ class ExtensionManager(
         }
     }
 
-    fun Extension.isBlacklisted(
+    fun Extension.isBlacklisted(blacklistEnabled: Boolean = preferences.enableSourceBlacklist().get()): Boolean {
-        blacklistEnabled: Boolean =
-            preferences.enableSourceBlacklist().get()
-    ): Boolean {
         return pkgName in BlacklistedSources.BLACKLISTED_EXTENSIONS && blacklistEnabled
     }
     // EXH <--

app/src/main/java/eu/kanade/tachiyomi/source/online/BrowseSourceFilterHeader.kt

@@ -2,7 +2,8 @@ package eu.kanade.tachiyomi.source.online
 
 import androidx.recyclerview.widget.RecyclerView
 import com.bluelinelabs.conductor.Controller
+import eu.kanade.tachiyomi.source.CatalogueSource
 
-interface BrowseSourceFilterHeader {
+interface BrowseSourceFilterHeader : CatalogueSource {
     fun getFilterHeader(controller: Controller): RecyclerView.Adapter<*>
 }

app/src/main/java/eu/kanade/tachiyomi/source/online/FollowsSource.kt

@@ -1,30 +1,29 @@
 package eu.kanade.tachiyomi.source.online
 
 import eu.kanade.tachiyomi.data.database.models.Track
+import eu.kanade.tachiyomi.source.CatalogueSource
 import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.source.model.SManga
 import exh.md.utils.FollowStatus
 import exh.metadata.metadata.base.RaisedSearchMetadata
-import kotlinx.coroutines.flow.Flow
-import rx.Observable
 
-interface FollowsSource {
+interface FollowsSource : CatalogueSource {
-    fun fetchFollows(): Observable<MangasPage>
+    suspend fun fetchFollows(): MangasPage
 
     /**
      * Returns a list of all Follows retrieved by Coroutines
      *
      * @param SManga all smanga found for user
      */
-    fun fetchAllFollows(forceHd: Boolean = false): Flow<List<Pair<SManga, RaisedSearchMetadata>>>
+    suspend fun fetchAllFollows(forceHd: Boolean = false): List<Pair<SManga, RaisedSearchMetadata>>
 
     /**
      * updates the follow status for a manga
      */
-    fun updateFollowStatus(mangaID: String, followStatus: FollowStatus): Flow<Boolean>
+    suspend fun updateFollowStatus(mangaID: String, followStatus: FollowStatus): Boolean
 
     /**
      * Get a MdList Track of the manga
      */
-    fun fetchTrackingInfo(url: String): Flow<Track>
+    suspend fun fetchTrackingInfo(url: String): Track
 }

app/src/main/java/eu/kanade/tachiyomi/source/online/LoginSource.kt

@@ -4,7 +4,7 @@ import android.app.Activity
 import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.ui.base.controller.DialogController
 
-interface LoginSource {
+interface LoginSource : Source {
     val needsLogin: Boolean
 
     fun isLogged(): Boolean

app/src/main/java/eu/kanade/tachiyomi/source/online/NamespaceSource.kt

@@ -1,3 +1,5 @@
 package eu.kanade.tachiyomi.source.online
 
-interface NamespaceSource
+import eu.kanade.tachiyomi.source.Source
+
+interface NamespaceSource : Source

app/src/main/java/eu/kanade/tachiyomi/source/online/RandomMangaSource.kt

@@ -1,7 +1,7 @@
 package eu.kanade.tachiyomi.source.online
 
-import kotlinx.coroutines.flow.Flow
+import tachiyomi.source.Source
 
-interface RandomMangaSource {
+interface RandomMangaSource : Source {
-    fun fetchRandomMangaUrl(): Flow<String>
+    suspend fun fetchRandomMangaUrl(): String
 }

app/src/main/java/eu/kanade/tachiyomi/source/online/SuspendHttpSource.kt

@@ -8,10 +8,10 @@ import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.source.model.SChapter
 import eu.kanade.tachiyomi.source.model.SManga
-import eu.kanade.tachiyomi.util.lang.asObservable
+import eu.kanade.tachiyomi.util.lang.runAsObservable
-import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
 import okhttp3.Request
 import okhttp3.Response
 import rx.Observable
@@ -29,15 +29,13 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param page the page number to retrieve.
      */
     final override fun fetchPopularManga(page: Int): Observable<MangasPage> {
-        return fetchPopularMangaFlow(page).asObservable()
+        return runAsObservable({ fetchPopularMangaSuspended(page) })
     }
 
-    open fun fetchPopularMangaFlow(page: Int): Flow<MangasPage> {
+    open suspend fun fetchPopularMangaSuspended(page: Int): MangasPage {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(popularMangaRequestSuspended(page)).await()
-            emit(
+            popularMangaParseSuspended(response)
-                popularMangaParseSuspended(response)
-            )
         }
     }
 
@@ -72,15 +70,13 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param filters the list of filters to apply.
      */
     final override fun fetchSearchManga(page: Int, query: String, filters: FilterList): Observable<MangasPage> {
-        return fetchSearchMangaSuspended(page, query, filters).asObservable()
+        return runAsObservable({ fetchSearchMangaSuspended(page, query, filters) })
     }
 
-    open fun fetchSearchMangaSuspended(page: Int, query: String, filters: FilterList): Flow<MangasPage> {
+    open suspend fun fetchSearchMangaSuspended(page: Int, query: String, filters: FilterList): MangasPage {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(searchMangaRequestSuspended(page, query, filters)).await()
-            emit(
+            searchMangaParseSuspended(response)
-                searchMangaParseSuspended(response)
-            )
         }
     }
 
@@ -114,15 +110,13 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param page the page number to retrieve.
      */
     final override fun fetchLatestUpdates(page: Int): Observable<MangasPage> {
-        return fetchLatestUpdatesFlow(page).asObservable()
+        return runAsObservable({ fetchLatestUpdatesSuspended(page) })
     }
 
-    open fun fetchLatestUpdatesFlow(page: Int): Flow<MangasPage> {
+    open suspend fun fetchLatestUpdatesSuspended(page: Int): MangasPage {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(latestUpdatesRequestSuspended(page)).await()
-            emit(
+            latestUpdatesParseSuspended(response)
-                latestUpdatesParseSuspended(response)
-            )
         }
     }
 
@@ -155,15 +149,13 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param manga the manga to be updated.
      */
     final override fun fetchMangaDetails(manga: SManga): Observable<SManga> {
-        return fetchMangaDetailsFlow(manga).asObservable()
+        return runAsObservable({ fetchMangaDetailsSuspended(manga) })
     }
 
-    open fun fetchMangaDetailsFlow(manga: SManga): Flow<SManga> {
+    open suspend fun fetchMangaDetailsSuspended(manga: SManga): SManga {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(mangaDetailsRequestSuspended(manga)).await()
-            emit(
+            mangaDetailsParseSuspended(response).apply { initialized = true }
-                mangaDetailsParseSuspended(response).apply { initialized = true }
-            )
         }
     }
 
@@ -200,20 +192,18 @@ abstract class SuspendHttpSource : HttpSource() {
      */
     final override fun fetchChapterList(manga: SManga): Observable<List<SChapter>> {
         return try {
-            fetchChapterListFlow(manga).asObservable()
+            runAsObservable({ fetchChapterListSuspended(manga) })
         } catch (e: LicencedException) {
             Observable.error(Exception("Licensed - No chapters to show"))
         }
     }
 
     @Throws(LicencedException::class)
-    open fun fetchChapterListFlow(manga: SManga): Flow<List<SChapter>> {
+    open suspend fun fetchChapterListSuspended(manga: SManga): List<SChapter> {
-        return flow {
+        return withContext(Dispatchers.IO) {
             if (manga.status != SManga.LICENSED) {
                 val response = client.newCall(chapterListRequestSuspended(manga)).await()
-                emit(
+                chapterListParseSuspended(response)
-                    chapterListParseSuspended(response)
-                )
             } else {
                 throw LicencedException("Licensed - No chapters to show")
             }
@@ -251,15 +241,13 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param chapter the chapter whose page list has to be fetched.
      */
     final override fun fetchPageList(chapter: SChapter): Observable<List<Page>> {
-        return fetchPageListFlow(chapter).asObservable()
+        return runAsObservable({ fetchPageListSuspended(chapter) })
     }
 
-    open fun fetchPageListFlow(chapter: SChapter): Flow<List<Page>> {
+    open suspend fun fetchPageListSuspended(chapter: SChapter): List<Page> {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(pageListRequestSuspended(chapter)).await()
-            emit(
+            pageListParseSuspended(response)
-                pageListParseSuspended(response)
-            )
         }
     }
 
@@ -295,15 +283,13 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param page the page whose source image has to be fetched.
      */
     final override fun fetchImageUrl(page: Page): Observable<String> {
-        return fetchImageUrlFlow(page).asObservable()
+        return runAsObservable({ fetchImageUrlSuspended(page) })
     }
 
-    open fun fetchImageUrlFlow(page: Page): Flow<String> {
+    open suspend fun fetchImageUrlSuspended(page: Page): String {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(imageUrlRequestSuspended(page)).await()
-            emit(
+            imageUrlParseSuspended(response)
-                imageUrlParseSuspended(response)
-            )
         }
     }
 
@@ -338,14 +324,12 @@ abstract class SuspendHttpSource : HttpSource() {
      * @param page the page whose source image has to be downloaded.
      */
     final override fun fetchImage(page: Page): Observable<Response> {
-        return fetchImageFlow(page).asObservable()
+        return runAsObservable({ fetchImageSuspended(page) })
     }
 
-    open fun fetchImageFlow(page: Page): Flow<Response> {
+    open suspend fun fetchImageSuspended(page: Page): Response {
-        return flow {
+        return withContext(Dispatchers.IO) {
-            emit(
+            client.newCallWithProgress(imageRequestSuspended(page), page).await()
-                client.newCallWithProgress(imageRequestSuspended(page), page).await()
-            )
         }
     }
 

app/src/main/java/eu/kanade/tachiyomi/source/online/UrlImportableSource.kt

@@ -13,7 +13,7 @@ interface UrlImportableSource : Source {
     }
 
     // This method is allowed to block for IO if necessary
-    fun mapUrlToMangaUrl(uri: Uri): String?
+    suspend fun mapUrlToMangaUrl(uri: Uri): String?
 
     fun cleanMangaUrl(url: String): String {
         return try {

app/src/main/java/eu/kanade/tachiyomi/source/online/all/EHentai.kt

@@ -910,7 +910,7 @@ class EHentai(
         "e-hentai.org"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         return when (uri.pathSegments.firstOrNull()) {
             "g" -> {
                 // Is already gallery page, do nothing

app/src/main/java/eu/kanade/tachiyomi/source/online/all/Hitomi.kt

@@ -128,7 +128,7 @@ class Hitomi(delegate: HttpSource, val context: Context) :
         "hitomi.la"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         val lcFirstPathSegment = uri.pathSegments.firstOrNull()?.toLowerCase() ?: return null
 
         if (lcFirstPathSegment != "manga" && lcFirstPathSegment != "reader") {

app/src/main/java/eu/kanade/tachiyomi/source/online/all/MangaDex.kt

@@ -48,7 +48,6 @@ import exh.ui.metadata.adapters.MangaDexDescriptionAdapter
 import exh.util.urlImportFetchSearchManga
 import exh.widget.preference.MangadexLoginDialog
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.withContext
 import okhttp3.CacheControl
@@ -101,7 +100,7 @@ class MangaDex(delegate: HttpSource, val context: Context) :
             }
         }
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         val lcFirstPathSegment = uri.pathSegments.firstOrNull()?.toLowerCase() ?: return null
 
         return if (lcFirstPathSegment == "title" || lcFirstPathSegment == "manga") {
@@ -152,7 +151,7 @@ class MangaDex(delegate: HttpSource, val context: Context) :
         ApiMangaParser(listOf(mdLang)).parseIntoMetadata(metadata, input, preferences.mangaDexForceLatestCovers().get())
     }
 
-    override fun fetchFollows(): Observable<MangasPage> {
+    override suspend fun fetchFollows(): MangasPage {
         return FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).fetchFollows()
     }
 
@@ -226,36 +225,36 @@ class MangaDex(delegate: HttpSource, val context: Context) :
         }
     }
 
-    override fun fetchAllFollows(forceHd: Boolean): Flow<List<Pair<SManga, MangaDexSearchMetadata>>> {
+    override suspend fun fetchAllFollows(forceHd: Boolean): List<Pair<SManga, MangaDexSearchMetadata>> {
-        return flow { emit(FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).fetchAllFollows(forceHd)) }
+        return withContext(Dispatchers.IO) { FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).fetchAllFollows(forceHd) }
     }
 
-    fun updateReadingProgress(track: Track): Flow<Boolean> {
+    suspend fun updateReadingProgress(track: Track): Boolean {
-        return flow { FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).updateReadingProgress(track) }
+        return withContext(Dispatchers.IO) { FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).updateReadingProgress(track) }
     }
 
-    fun updateRating(track: Track): Flow<Boolean> {
+    suspend fun updateRating(track: Track): Boolean {
-        return flow { FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).updateRating(track) }
+        return withContext(Dispatchers.IO) { FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).updateRating(track) }
     }
 
-    override fun fetchTrackingInfo(url: String): Flow<Track> {
+    override suspend fun fetchTrackingInfo(url: String): Track {
-        return flow {
+        return withContext(Dispatchers.IO) {
             if (!isLogged()) {
                 throw Exception("Not Logged in")
             }
-            emit(FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).fetchTrackingInfo(url))
+            FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).fetchTrackingInfo(url)
         }
     }
 
-    override fun updateFollowStatus(mangaID: String, followStatus: FollowStatus): Flow<Boolean> {
+    override suspend fun updateFollowStatus(mangaID: String, followStatus: FollowStatus): Boolean {
-        return flow { emit(FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).updateFollowStatus(mangaID, followStatus)) }
+        return withContext(Dispatchers.IO) { FollowsHandler(client, headers, Injekt.get(), useLowQualityThumbnail()).updateFollowStatus(mangaID, followStatus) }
     }
 
     override fun getFilterHeader(controller: Controller): MangaDexFabHeaderAdapter {
         return MangaDexFabHeaderAdapter(controller, this)
     }
 
-    override fun fetchRandomMangaUrl(): Flow<String> {
+    override suspend fun fetchRandomMangaUrl(): String {
         return MangaHandler(client, headers, listOf(mdLang)).fetchRandomMangaId()
     }
 

app/src/main/java/eu/kanade/tachiyomi/source/online/all/MergedSource.kt

@@ -14,7 +14,6 @@ import eu.kanade.tachiyomi.source.model.SChapter
 import eu.kanade.tachiyomi.source.model.SManga
 import eu.kanade.tachiyomi.source.online.SuspendHttpSource
 import eu.kanade.tachiyomi.util.chapter.syncChaptersWithSource
-import eu.kanade.tachiyomi.util.lang.asFlow
 import eu.kanade.tachiyomi.util.lang.await
 import eu.kanade.tachiyomi.util.lang.awaitSingle
 import eu.kanade.tachiyomi.util.lang.awaitSingleOrNull
@@ -22,13 +21,6 @@ import eu.kanade.tachiyomi.util.shouldDownloadNewChapters
 import exh.MERGED_SOURCE_ID
 import exh.merged.sql.models.MergedMangaReference
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.flow.buffer
-import kotlinx.coroutines.flow.collect
-import kotlinx.coroutines.flow.emptyFlow
-import kotlinx.coroutines.flow.flatMapMerge
-import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.runBlocking
@@ -57,15 +49,15 @@ class MergedSource : SuspendHttpSource() {
     override suspend fun chapterListParseSuspended(response: Response) = throw UnsupportedOperationException()
     override suspend fun pageListParseSuspended(response: Response) = throw UnsupportedOperationException()
     override suspend fun imageUrlParseSuspended(response: Response) = throw UnsupportedOperationException()
-    override fun fetchChapterListFlow(manga: SManga) = throw UnsupportedOperationException()
+    override suspend fun fetchChapterListSuspended(manga: SManga) = throw UnsupportedOperationException()
-    override fun fetchImageFlow(page: Page) = throw UnsupportedOperationException()
+    override suspend fun fetchImageSuspended(page: Page) = throw UnsupportedOperationException()
-    override fun fetchImageUrlFlow(page: Page) = throw UnsupportedOperationException()
+    override suspend fun fetchImageUrlSuspended(page: Page) = throw UnsupportedOperationException()
-    override fun fetchPageListFlow(chapter: SChapter) = throw UnsupportedOperationException()
+    override suspend fun fetchPageListSuspended(chapter: SChapter) = throw UnsupportedOperationException()
-    override fun fetchLatestUpdatesFlow(page: Int) = throw UnsupportedOperationException()
+    override suspend fun fetchLatestUpdatesSuspended(page: Int) = throw UnsupportedOperationException()
-    override fun fetchPopularMangaFlow(page: Int) = throw UnsupportedOperationException()
+    override suspend fun fetchPopularMangaSuspended(page: Int) = throw UnsupportedOperationException()
 
-    override fun fetchMangaDetailsFlow(manga: SManga): Flow<SManga> {
+    override suspend fun fetchMangaDetailsSuspended(manga: SManga): SManga {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val mergedManga = db.getManga(manga.url, id).await() ?: throw Exception("merged manga not in db")
             val mangaReferences = mergedManga.id?.let { withContext(Dispatchers.IO) { db.getMergedMangaReferences(it).await() } } ?: throw Exception("merged manga id is null")
             if (mangaReferences.isEmpty()) throw IllegalArgumentException("Manga references are empty, info unavailable, merge is likely corrupted")
@@ -75,14 +67,12 @@ class MergedSource : SuspendHttpSource() {
             }
             ) throw IllegalArgumentException("Manga references contain only the merged reference, merge is likely corrupted")
 
-            emit(
+            SManga.create().apply {
-                SManga.create().apply {
+                val mangaInfoReference = mangaReferences.firstOrNull { it.isInfoManga } ?: mangaReferences.firstOrNull { it.mangaId != it.mergeId }
-                    val mangaInfoReference = mangaReferences.firstOrNull { it.isInfoManga } ?: mangaReferences.firstOrNull { it.mangaId != it.mergeId }
+                val dbManga = mangaInfoReference?.let { withContext(Dispatchers.IO) { db.getManga(it.mangaUrl, it.mangaSourceId).await() } }
-                    val dbManga = mangaInfoReference?.let { withContext(Dispatchers.IO) { db.getManga(it.mangaUrl, it.mangaSourceId).await() } }
+                this.copyFrom(dbManga ?: mergedManga)
-                    this.copyFrom(dbManga ?: mergedManga)
+                url = manga.url
-                    url = manga.url
+            }
-                }
-            )
         }
     }
 
@@ -131,39 +121,45 @@ class MergedSource : SuspendHttpSource() {
         return chapterList.maxByOrNull { it.chapter_number }?.manga_id
     }
 
-    fun fetchChaptersForMergedManga(manga: Manga, downloadChapters: Boolean = true, editScanlators: Boolean = false, dedupe: Boolean = true): Flow<List<Chapter>> {
+    suspend fun fetchChaptersForMergedManga(manga: Manga, downloadChapters: Boolean = true, editScanlators: Boolean = false, dedupe: Boolean = true): List<Chapter> {
-        return flow {
+        return withContext(Dispatchers.IO) {
-            withContext(Dispatchers.IO) {
+            fetchChaptersAndSync(manga, downloadChapters)
-                fetchChaptersAndSync(manga, downloadChapters).collect()
+            getChaptersFromDB(manga, editScanlators, dedupe).awaitSingleOrNull() ?: emptyList()
-            }
-            emit(
-                getChaptersFromDB(manga, editScanlators, dedupe).awaitSingleOrNull() ?: emptyList<Chapter>()
-            )
         }
     }
 
-    suspend fun fetchChaptersAndSync(manga: Manga, downloadChapters: Boolean = true): Flow<Pair<List<Chapter>, List<Chapter>>> {
+    suspend fun fetchChaptersAndSync(manga: Manga, downloadChapters: Boolean = true): Pair<List<Chapter>, List<Chapter>> {
         val mangaReferences = db.getMergedMangaReferences(manga.id!!).await()
         if (mangaReferences.isEmpty()) throw IllegalArgumentException("Manga references are empty, chapters unavailable, merge is likely corrupted")
 
         val ifDownloadNewChapters = downloadChapters && manga.shouldDownloadNewChapters(db, preferences)
-        return mangaReferences.filter { it.mangaSourceId != MERGED_SOURCE_ID }.asFlow().map {
+        return mangaReferences.filter { it.mangaSourceId != MERGED_SOURCE_ID }.map {
             load(db, sourceManager, it)
-        }.buffer().flatMapMerge { loadedManga ->
+        }.mapNotNull { loadedManga ->
             withContext(Dispatchers.IO) {
                 if (loadedManga.manga != null && loadedManga.reference.getChapterUpdates) {
-                    loadedManga.source.fetchChapterList(loadedManga.manga).asFlow()
+                    loadedManga.source.fetchChapterList(loadedManga.manga).awaitSingle()
-                        .map { syncChaptersWithSource(db, it, loadedManga.manga, loadedManga.source) }
+                        .let { syncChaptersWithSource(db, it, loadedManga.manga, loadedManga.source) }
-                        .onEach {
+                        .also {
                             if (ifDownloadNewChapters && loadedManga.reference.downloadChapters) {
                                 downloadManager.downloadChapters(loadedManga.manga, it.first)
                             }
                         }
                 } else {
-                    emptyFlow()
+                    null
                 }
             }
-        }.buffer()
+        }.let { pairs ->
+            val firsts = mutableListOf<Chapter>()
+            val seconds = mutableListOf<Chapter>()
+
+            pairs.forEach {
+                firsts.addAll(it.first)
+                seconds.addAll(it.second)
+            }
+
+            firsts to seconds
+        }
     }
 
     suspend fun load(db: DatabaseHelper, sourceManager: SourceManager, reference: MergedMangaReference): LoadedMangaSource {

app/src/main/java/eu/kanade/tachiyomi/source/online/all/NHentai.kt

@@ -165,7 +165,7 @@ class NHentai(delegate: HttpSource, val context: Context) :
         "nhentai.net"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         if (uri.pathSegments.firstOrNull()?.toLowerCase() != "g") {
             return null
         }

app/src/main/java/eu/kanade/tachiyomi/source/online/all/PervEden.kt

@@ -129,7 +129,7 @@ class PervEden(delegate: HttpSource, val context: Context) :
         }
     }
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         val newUri = "http://www.perveden.com/".toUri().buildUpon()
         uri.pathSegments.take(3).forEach {
             newUri.appendPath(it)

app/src/main/java/eu/kanade/tachiyomi/source/online/english/EightMuses.kt

@@ -93,7 +93,7 @@ class EightMuses(delegate: HttpSource, val context: Context) :
         "8muses.com"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         var path = uri.pathSegments.drop(2)
         if (uri.pathSegments[1].toLowerCase() == "picture") {
             path = path.dropLast(1)

app/src/main/java/eu/kanade/tachiyomi/source/online/english/HBrowse.kt

@@ -83,7 +83,7 @@ class HBrowse(delegate: HttpSource, val context: Context) :
         "hbrowse.com"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         return "/${uri.pathSegments.first()}/c00001/"
     }
 

app/src/main/java/eu/kanade/tachiyomi/source/online/english/HentaiCafe.kt

@@ -105,7 +105,7 @@ class HentaiCafe(delegate: HttpSource, val context: Context) :
         "hentai.cafe"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         val lcFirstPathSegment = uri.pathSegments.firstOrNull()?.takeUnless { it.equals("manga", true) } ?: return null
 
         return if (lcFirstPathSegment.equals("hc.fyi", true)) {

app/src/main/java/eu/kanade/tachiyomi/source/online/english/Pururin.kt

@@ -114,7 +114,7 @@ class Pururin(delegate: HttpSource, val context: Context) :
         "www.pururin.io"
     )
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         return "${PururinSearchMetadata.BASE_URL}/gallery/${uri.pathSegments[1]}/${uri.lastPathSegment}"
     }
 

app/src/main/java/eu/kanade/tachiyomi/source/online/english/Tsumino.kt

@@ -40,7 +40,7 @@ class Tsumino(delegate: HttpSource, val context: Context) :
             super.fetchSearchManga(page, query, filters)
         }
 
-    override fun mapUrlToMangaUrl(uri: Uri): String? {
+    override suspend fun mapUrlToMangaUrl(uri: Uri): String? {
         val lcFirstPathSegment = uri.pathSegments.firstOrNull()?.toLowerCase(Locale.ROOT) ?: return null
         if (lcFirstPathSegment != "read" && lcFirstPathSegment != "book" && lcFirstPathSegment != "entry") {
             return null

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

@@ -34,7 +34,6 @@ import exh.md.utils.FollowStatus
 import exh.md.utils.MdUtil
 import exh.util.isLewd
 import exh.util.nullIfBlank
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.runBlocking
 import rx.Observable
 import rx.Subscription
@@ -543,7 +542,7 @@ class LibraryPresenter(
         launchIO {
             MdUtil.getEnabledMangaDex(preferences, sourceManager)?.let { mdex ->
                 mangaList.forEach {
-                    mdex.updateFollowStatus(MdUtil.getMangaId(it.url), FollowStatus.READING).collect()
+                    mdex.updateFollowStatus(MdUtil.getMangaId(it.url), FollowStatus.READING)
                 }
             }
         }

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

@@ -29,10 +29,10 @@ import eu.kanade.tachiyomi.ui.manga.chapter.ChapterItem
 import eu.kanade.tachiyomi.util.chapter.ChapterSettingsHelper
 import eu.kanade.tachiyomi.util.chapter.syncChaptersWithSource
 import eu.kanade.tachiyomi.util.isLocal
-import eu.kanade.tachiyomi.util.lang.asObservable
 import eu.kanade.tachiyomi.util.lang.await
 import eu.kanade.tachiyomi.util.lang.isNullOrUnsubscribed
 import eu.kanade.tachiyomi.util.lang.launchIO
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import eu.kanade.tachiyomi.util.prepUpdateCover
 import eu.kanade.tachiyomi.util.removeCovers
 import eu.kanade.tachiyomi.util.shouldDownloadNewChapters
@@ -766,7 +766,7 @@ class MangaPresenter(
                 )
             // SY -->
         } else {
-            Observable.defer { source.fetchChaptersForMergedManga(manga, manualFetch, true, dedupe).asObservable() }
+            Observable.defer { runAsObservable({ source.fetchChaptersForMergedManga(manga, manualFetch, true, dedupe) }) }
                 .subscribeOn(Schedulers.io())
                 .observeOn(AndroidSchedulers.mainThread())
                 .doOnNext {

app/src/main/java/eu/kanade/tachiyomi/ui/reader/ReaderPresenter.kt

@@ -45,12 +45,11 @@ import exh.metadata.metadata.base.getFlatMetadataForManga
 import exh.source.getMainSource
 import exh.util.defaultReaderType
 import exh.util.shouldDeleteChapters
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
-import rx.Completable
 import rx.Observable
 import rx.Subscription
 import rx.android.schedulers.AndroidSchedulers

app/src/main/java/exh/GalleryAdder.kt

@@ -80,7 +80,7 @@ class GalleryAdder {
             }
 
             // Map URL to manga URL
-            val realUrl = try {
+            val realMangaUrl = try {
                 source.mapUrlToMangaUrl(uri)
             } catch (e: Exception) {
                 logger.e(context.getString(R.string.gallery_adder_uri_map_to_manga_error), e)
@@ -88,18 +88,18 @@ class GalleryAdder {
             } ?: return GalleryAddEvent.Fail.UnknownType(url, context)
 
             // Clean URL
-            val cleanedUrl = try {
+            val cleanedMangaUrl = try {
-                source.cleanMangaUrl(realUrl)
+                source.cleanMangaUrl(realMangaUrl)
             } catch (e: Exception) {
                 logger.e(context.getString(R.string.gallery_adder_uri_clean_error), e)
                 null
             } ?: return GalleryAddEvent.Fail.UnknownType(url, context)
 
             // Use manga in DB if possible, otherwise, make a new manga
-            val manga = db.getManga(cleanedUrl, source.id).executeOnIO()
+            val manga = db.getManga(cleanedMangaUrl, source.id).executeOnIO()
                 ?: Manga.create(source.id).apply {
-                    this.url = cleanedUrl
+                    this.url = cleanedMangaUrl
-                    title = realUrl
+                    title = realMangaUrl
                 }
 
             // Insert created manga if not in DB before fetching details
@@ -163,7 +163,8 @@ sealed class GalleryAddEvent {
     class Success(
         override val galleryUrl: String,
         val manga: Manga,
-        val context: Context
+        val context: Context,
+        val chapter: Chapter? = null
     ) : GalleryAddEvent() {
         override val galleryTitle = manga.title
         override val logMessage = context.getString(R.string.batch_add_success_log_message, galleryTitle)

app/src/main/java/exh/md/MangaDexFabHeaderAdapter.kt

@@ -16,7 +16,6 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.singleOrNull
 import reactivecircus.flowbinding.android.view.clicks
 
 class MangaDexFabHeaderAdapter(val controller: Controller, val source: CatalogueSource) :
@@ -46,7 +45,7 @@ class MangaDexFabHeaderAdapter(val controller: Controller, val source: Catalogue
                 .launchIn(scope)
             binding.mangadexRandom.clicks()
                 .onEach {
-                    (source as? RandomMangaSource)?.fetchRandomMangaUrl()?.singleOrNull()?.let { randomMangaId ->
+                    (source as? RandomMangaSource)?.fetchRandomMangaUrl()?.let { randomMangaId ->
                         controller.router.replaceTopController(BrowseSourceController(source, randomMangaId).withFadeTransaction())
                     }
                 }

app/src/main/java/exh/md/follows/MangaDexFollowsPager.kt

@@ -3,6 +3,7 @@ package exh.md.follows
 import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.source.online.all.MangaDex
 import eu.kanade.tachiyomi.ui.browse.source.browse.Pager
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import rx.Observable
 import rx.android.schedulers.AndroidSchedulers
 import rx.schedulers.Schedulers
@@ -13,7 +14,7 @@ import rx.schedulers.Schedulers
 class MangaDexFollowsPager(val source: MangaDex) : Pager() {
 
     override fun requestNext(): Observable<MangasPage> {
-        return source.fetchFollows()
+        return runAsObservable({ source.fetchFollows() })
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
             .doOnNext { onPageReceived(it) }

app/src/main/java/exh/md/handlers/FollowsHandler.kt

@@ -6,7 +6,6 @@ import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.data.track.TrackManager
 import eu.kanade.tachiyomi.network.GET
 import eu.kanade.tachiyomi.network.POST
-import eu.kanade.tachiyomi.network.asObservable
 import eu.kanade.tachiyomi.network.await
 import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.source.model.MetadataMangasPage
@@ -26,17 +25,16 @@ import okhttp3.Headers
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
-import rx.Observable
 
 class FollowsHandler(val client: OkHttpClient, val headers: Headers, val preferences: PreferencesHelper, private val useLowQualityCovers: Boolean) {
 
     /**
      * fetch follows by page
      */
-    fun fetchFollows(): Observable<MangasPage> {
+    suspend fun fetchFollows(): MangasPage {
         return client.newCall(followsListRequest())
-            .asObservable()
+            .await()
-            .map { response ->
+            .let { response ->
                 followsParseMangaPage(response)
             }
     }

app/src/main/java/exh/md/handlers/MangaHandler.kt

@@ -8,8 +8,6 @@ import eu.kanade.tachiyomi.source.model.SChapter
 import eu.kanade.tachiyomi.source.model.SManga
 import exh.md.utils.MdUtil
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.withContext
 import okhttp3.CacheControl
 import okhttp3.Headers
@@ -95,10 +93,10 @@ class MangaHandler(val client: OkHttpClient, val headers: Headers, val langs: Li
             }
     }
 
-    fun fetchRandomMangaId(): Flow<String> {
+    suspend fun fetchRandomMangaId(): String {
-        return flow {
+        return withContext(Dispatchers.IO) {
             val response = client.newCall(randomMangaRequest()).await()
-            emit(ApiMangaParser(langs).randomMangaIdParse(response))
+            ApiMangaParser(langs).randomMangaIdParse(response)
         }
     }