Use coroutine job for fetching next source page

(cherry picked from commit e6f3cd03bb)

# Conflicts:
#	app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/browse/BrowseSourcePresenter.kt

app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/browse/BrowseSourcePresenter.kt

@@ -41,6 +41,7 @@ import eu.kanade.tachiyomi.util.lang.withUIContext
 import eu.kanade.tachiyomi.util.removeCovers
 import exh.savedsearches.EXHSavedSearch
 import exh.savedsearches.JsonSavedSearch
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.catch
@@ -52,7 +53,6 @@ import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
-import rx.Observable
 import rx.Subscription
 import rx.android.schedulers.AndroidSchedulers
 import rx.schedulers.Schedulers
@@ -117,7 +117,7 @@ open class BrowseSourcePresenter(
     /**
      * Subscription for one request from the pager.
      */
-    private var pageSubscription: Subscription? = null
+    private var nextPageJob: Job? = null
 
     private val loggedServices by lazy { Injekt.get<TrackManager>().services.filter { it.isLogged } }
 
@@ -204,14 +204,14 @@ open class BrowseSourcePresenter(
     fun requestNext() {
         if (!hasNextPage()) return
 
-        pageSubscription?.let { remove(it) }
-        pageSubscription = Observable.defer { pager.requestNext() }
-            .subscribeFirst(
-                { _, _ ->
-                    // Nothing to do when onNext is emitted.
-                },
-                BrowseSourceController::onAddPageError
-            )
+        nextPageJob?.cancel()
+        nextPageJob = launchIO {
+            try {
+                pager.requestNextPage()
+            } catch (e: Throwable) {
+                view?.onAddPageError(e)
+            }
+        }
     }
 
     /**

app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/browse/Pager.kt

@@ -21,7 +21,7 @@ abstract class Pager(var currentPage: Int = 1) {
         return results.asObservable()
     }
 
-    abstract fun requestNext(): Observable<MangasPage>
+    abstract suspend fun requestNextPage()
 
     fun onPageReceived(mangasPage: MangasPage) {
         val page = currentPage

app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/browse/SourcePager.kt

@@ -2,14 +2,11 @@ package eu.kanade.tachiyomi.ui.browse.source.browse
 
 import eu.kanade.tachiyomi.source.CatalogueSource
 import eu.kanade.tachiyomi.source.model.FilterList
-import eu.kanade.tachiyomi.source.model.MangasPage
-import rx.Observable
-import rx.android.schedulers.AndroidSchedulers
-import rx.schedulers.Schedulers
+import eu.kanade.tachiyomi.util.lang.awaitSingle
 
 open class SourcePager(val source: CatalogueSource, val query: String, val filters: FilterList) : Pager() {
 
-    override fun requestNext(): Observable<MangasPage> {
+    override suspend fun requestNextPage() {
         val page = currentPage
 
         val observable = if (query.isBlank() && filters.isEmpty()) {
@@ -18,15 +15,12 @@ open class SourcePager(val source: CatalogueSource, val query: String, val filte
             source.fetchSearchManga(page, query, filters)
         }
 
-        return observable
-            .subscribeOn(Schedulers.io())
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnNext {
-                if (it.mangas.isNotEmpty()) {
-                    onPageReceived(it)
-                } else {
-                    throw NoResultsException()
-                }
-            }
+        val mangasPage = observable.awaitSingle()
+
+        if (mangasPage.mangas.isNotEmpty()) {
+            onPageReceived(mangasPage)
+        } else {
+            throw NoResultsException()
+        }
     }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/latest/LatestUpdatesPager.kt

@@ -1,21 +1,13 @@
 package eu.kanade.tachiyomi.ui.browse.source.latest
 
 import eu.kanade.tachiyomi.source.CatalogueSource
-import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.ui.browse.source.browse.Pager
-import rx.Observable
-import rx.android.schedulers.AndroidSchedulers
-import rx.schedulers.Schedulers
+import eu.kanade.tachiyomi.util.lang.awaitSingle
 
-/**
- * LatestUpdatesPager inherited from the general Pager.
- */
 class LatestUpdatesPager(val source: CatalogueSource) : Pager() {
 
-    override fun requestNext(): Observable<MangasPage> {
-        return source.fetchLatestUpdates(currentPage)
-            .subscribeOn(Schedulers.io())
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnNext { onPageReceived(it) }
+    override suspend fun requestNextPage() {
+        val mangasPage = source.fetchLatestUpdates(currentPage).awaitSingle()
+        onPageReceived(mangasPage)
     }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/latest/LatestUpdatesPresenter.kt

@@ -4,9 +4,6 @@ import eu.kanade.tachiyomi.source.model.FilterList
 import eu.kanade.tachiyomi.ui.browse.source.browse.BrowseSourcePresenter
 import eu.kanade.tachiyomi.ui.browse.source.browse.Pager
 
-/**
- * Presenter of [LatestUpdatesController]. Inherit BrowseCataloguePresenter.
- */
 class LatestUpdatesPresenter(sourceId: Long) : BrowseSourcePresenter(sourceId) {
 
     override fun createPager(query: String, filters: FilterList): Pager {