Refactor bridged RxJava/coroutine calls in MangaPresenter

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

@@ -25,14 +25,14 @@ 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.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.lang.launchUI
 import eu.kanade.tachiyomi.util.prepUpdateCover
 import eu.kanade.tachiyomi.util.removeCovers
 import eu.kanade.tachiyomi.util.shouldDownloadNewChapters
 import eu.kanade.tachiyomi.util.updateCoverLastModified
 import eu.kanade.tachiyomi.widget.ExtendedNavigationView.Item.TriStateGroup.State
+import kotlinx.coroutines.Job
 import rx.Observable
 import rx.Subscription
 import rx.android.schedulers.AndroidSchedulers
@@ -55,7 +55,7 @@ class MangaPresenter(
     /**
      * Subscription to update the manga from the source.
      */
-    private var fetchMangaSubscription: Subscription? = null
+    private var fetchMangaJob: Job? = null
 
     /**
      * List of chapters of the manga. It's always unfiltered and unsorted.
@@ -79,7 +79,7 @@ class MangaPresenter(
     /**
      * Subscription to retrieve the new list of chapters from the source.
      */
-    private var fetchChaptersSubscription: Subscription? = null
+    private var fetchChaptersJob: Job? = null
 
     /**
      * Subscription to observe download status changes.
@@ -160,26 +160,21 @@ class MangaPresenter(
      * Fetch manga information from source.
      */
     fun fetchMangaFromSource(manualFetch: Boolean = false) {
-        if (!fetchMangaSubscription.isNullOrUnsubscribed()) return
-        fetchMangaSubscription = Observable.defer {
-            runAsObservable({
+        if (fetchMangaJob?.isActive == true) return
+        fetchMangaJob = launchIO {
+            try {
                 val networkManga = source.getMangaDetails(manga.toMangaInfo())
                 val sManga = networkManga.toSManga()
                 manga.prepUpdateCover(coverCache, sManga, manualFetch)
                 manga.copyFrom(sManga)
                 manga.initialized = true
-                db.insertManga(manga).executeAsBlocking()
-                manga
-            })
+                db.insertManga(manga).await()
+
+                launchUI { view?.onFetchMangaInfoDone() }
+            } catch (e: Throwable) {
+                launchUI { view?.onFetchMangaInfoError(e) }
+            }
         }
-            .subscribeOn(Schedulers.io())
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribeFirst(
-                { view, _ ->
-                    view.onFetchMangaInfoDone()
-                },
-                MangaController::onFetchMangaInfoError
-            )
     }
 
     /**
@@ -354,27 +349,22 @@ class MangaPresenter(
     fun fetchChaptersFromSource(manualFetch: Boolean = false) {
         hasRequested = true
 
-        if (!fetchChaptersSubscription.isNullOrUnsubscribed()) return
-        fetchChaptersSubscription = Observable.defer {
-            runAsObservable({
-                source.getChapterList(manga.toMangaInfo())
+        if (fetchChaptersJob?.isActive == true) return
+        fetchChaptersJob = launchIO {
+            try {
+                val chapters = source.getChapterList(manga.toMangaInfo())
                     .map { it.toSChapter() }
-            })
-        }
-            .subscribeOn(Schedulers.io())
-            .map { syncChaptersWithSource(db, it, manga, source) }
-            .doOnNext {
+
+                val (newChapters, _) = syncChaptersWithSource(db, chapters, manga, source)
                 if (manualFetch) {
-                    downloadNewChapters(it.first)
+                    downloadNewChapters(newChapters)
                 }
+
+                launchUI { view?.onFetchChaptersDone() }
+            } catch (e: Throwable) {
+                view?.onFetchChaptersError(e)
             }
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribeFirst(
-                { view, _ ->
-                    view.onFetchChaptersDone()
-                },
-                MangaController::onFetchChaptersError
-            )
+        }
     }
 
     /**