Replace PageLoader.getPage() with PageLoader.loadPage() (#8976)

* Follow page status via StateFlow

Keep getPage subscription since it's needed to load the pages

* Replace PageLoader.getPage with PageLoader.loadPage

(cherry picked from commit 2ef1f07aaea0852c13a4eb4096ac96c8aa507c39)

# Conflicts:
#	app/src/main/java/eu/kanade/tachiyomi/ui/reader/loader/HttpPageLoader.kt
#	app/src/main/java/eu/kanade/tachiyomi/ui/reader/viewer/pager/PagerPageHolder.kt

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

@@ -4,7 +4,6 @@ import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import eu.kanade.tachiyomi.util.lang.compareToCaseInsensitiveNaturalOrder
 import eu.kanade.tachiyomi.util.system.ImageUtil
-import rx.Observable
 import java.io.File
 import java.io.FileInputStream
 
@@ -30,9 +29,7 @@ class DirectoryPageLoader(val file: File) : PageLoader() {
     }
 
     /**
-     * Returns an observable that emits a ready state.
+     * No additional action required to load the page
      */
-    override fun getPage(page: ReaderPage): Observable<Page.State> {
+    override suspend fun loadPage(page: ReaderPage) {}
-        return Observable.just(Page.State.READY)
-    }
 }

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

@@ -10,7 +10,6 @@ import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.ui.reader.model.ReaderChapter
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
-import rx.Observable
 import tachiyomi.domain.manga.model.Manga
 import uy.kohesive.injekt.injectLazy
 import java.io.File
@@ -65,7 +64,7 @@ class DownloadPageLoader(
         }
     }
 
-    override fun getPage(page: ReaderPage): Observable<Page.State> {
+    override suspend fun loadPage(page: ReaderPage) {
-        return zipPageLoader?.getPage(page) ?: Observable.just(Page.State.READY)
+        zipPageLoader?.loadPage(page)
     }
 }

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

@@ -3,7 +3,6 @@ package eu.kanade.tachiyomi.ui.reader.loader
 import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import eu.kanade.tachiyomi.util.storage.EpubFile
-import rx.Observable
 import java.io.File
 
 /**
@@ -39,15 +38,9 @@ class EpubPageLoader(file: File) : PageLoader() {
     }
 
     /**
-     * Returns an observable that emits a ready state unless the loader was recycled.
+     * No additional action required to load the page
      */
-    override fun getPage(page: ReaderPage): Observable<Page.State> {
+    override suspend fun loadPage(page: ReaderPage) {
-        return Observable.just(
+        check(!isRecycled)
-            if (isRecycled) {
-                Page.State.ERROR
-            } else {
-                Page.State.READY
-            },
-        )
     }
 }

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

@@ -10,6 +10,7 @@ import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import eu.kanade.tachiyomi.ui.reader.setting.ReaderPreferences
 import eu.kanade.tachiyomi.util.lang.awaitSingle
 import eu.kanade.tachiyomi.util.lang.launchIO
+import eu.kanade.tachiyomi.util.lang.withIOContext
 import exh.source.isEhBasedSource
 import exh.util.DataSaver
 import exh.util.DataSaver.Companion.fetchImage
@@ -21,10 +22,7 @@ import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.runInterruptible
-import rx.Observable
+import kotlinx.coroutines.suspendCancellableCoroutine
-import rx.schedulers.Schedulers
-import rx.subjects.PublishSubject
-import rx.subjects.SerializedSubject
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 import java.util.concurrent.PriorityBlockingQueue
@@ -69,7 +67,7 @@ class HttpPageLoader(
                 }
                     .filter { it.status == Page.State.QUEUE }
                     .collect {
-                        loadPage(it)
+                        _loadPage(it)
                     }
             }
             // EXH -->
@@ -132,11 +130,10 @@ class HttpPageLoader(
     }
 
     /**
-     * Returns an observable that loads a page through the queue and listens to its result to
+     * Loads a page through the queue. Handles re-enqueueing pages if they were evicted from the cache.
-     * emit new states. It handles re-enqueueing pages if they were evicted from the cache.
      */
-    override fun getPage(page: ReaderPage): Observable<Page.State> {
+    override suspend fun loadPage(page: ReaderPage) {
-        return Observable.defer {
+        withIOContext {
             val imageUrl = page.imageUrl
 
             // Check if the image has been deleted
@@ -149,17 +146,14 @@ class HttpPageLoader(
                 page.status = Page.State.QUEUE
             }
 
-            val statusSubject = SerializedSubject(PublishSubject.create<Page.State>())
-            page.statusSubject = statusSubject
-
             val queuedPages = mutableListOf<PriorityPage>()
             if (page.status == Page.State.QUEUE) {
                 queuedPages += PriorityPage(page, 1).also { queue.offer(it) }
             }
             queuedPages += preloadNextPages(page, preloadSize)
 
-            statusSubject.startWith(page.status)
+            suspendCancellableCoroutine<Nothing> { continuation ->
-                .doOnUnsubscribe {
+                continuation.invokeOnCancellation {
                     queuedPages.forEach {
                         if (it.page.status == Page.State.QUEUE) {
                             queue.remove(it)
@@ -167,8 +161,7 @@ class HttpPageLoader(
                     }
                 }
             }
-            .subscribeOn(Schedulers.io())
+        }
-            .unsubscribeOn(Schedulers.io())
     }
 
     /**
@@ -238,7 +231,7 @@ class HttpPageLoader(
      *
      * @param page the page whose source image has to be downloaded.
      */
-    private suspend fun loadPage(page: ReaderPage) {
+    private suspend fun _loadPage(page: ReaderPage) {
         try {
             if (page.imageUrl.isNullOrEmpty()) {
                 page.status = Page.State.LOAD_PAGE

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

@@ -1,9 +1,7 @@
 package eu.kanade.tachiyomi.ui.reader.loader
 
 import androidx.annotation.CallSuper
-import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
-import rx.Observable
 
 /**
  * A loader used to load pages into the reader. Any open resources must be cleaned up when the
@@ -32,9 +30,11 @@ abstract class PageLoader {
     abstract suspend fun getPages(): List<ReaderPage>
 
     /**
-     * Returns an observable that should inform of the progress of the page
+     * Loads the page. May also preload other pages.
+     * Progress of the page loading should be followed via [page.statusFlow].
+     * [loadPage] is not currently guaranteed to complete, so it should be launched asynchronously.
      */
-    abstract fun getPage(page: ReaderPage): Observable<Page.State>
+    abstract suspend fun loadPage(page: ReaderPage)
 
     /**
      * Retries the given [page] in case it failed to load. This method only makes sense when an

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

@@ -6,7 +6,6 @@ import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import eu.kanade.tachiyomi.util.lang.compareToCaseInsensitiveNaturalOrder
 import eu.kanade.tachiyomi.util.system.ImageUtil
-import rx.Observable
 import java.io.File
 import java.io.InputStream
 import java.io.PipedInputStream
@@ -55,16 +54,10 @@ class RarPageLoader(file: File) : PageLoader() {
     }
 
     /**
-     * Returns an observable that emits a ready state unless the loader was recycled.
+     * No additional action required to load the page
      */
-    override fun getPage(page: ReaderPage): Observable<Page.State> {
+    override suspend fun loadPage(page: ReaderPage) {
-        return Observable.just(
+        check(!isRecycled)
-            if (isRecycled) {
-                Page.State.ERROR
-            } else {
-                Page.State.READY
-            },
-        )
     }
 
     /**

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

@@ -5,7 +5,6 @@ import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import eu.kanade.tachiyomi.util.lang.compareToCaseInsensitiveNaturalOrder
 import eu.kanade.tachiyomi.util.system.ImageUtil
-import rx.Observable
 import java.io.File
 import java.nio.charset.StandardCharsets
 import java.util.zip.ZipFile
@@ -49,15 +48,9 @@ class ZipPageLoader(file: File) : PageLoader() {
     }
 
     /**
-     * Returns an observable that emits a ready state unless the loader was recycled.
+     * No additional action required to load the page
      */
-    override fun getPage(page: ReaderPage): Observable<Page.State> {
+    override suspend fun loadPage(page: ReaderPage) {
-        return Observable.just(
+        check(!isRecycled)
-            if (isRecycled) {
-                Page.State.ERROR
-            } else {
-                Page.State.READY
-            },
-        )
     }
 }

app/src/main/java/eu/kanade/tachiyomi/ui/reader/viewer/pager/PagerPageHolder.kt

@@ -67,9 +67,14 @@ class PagerPageHolder(
     private val scope = MainScope()
 
     /**
-     * Subscription for status changes of the page.
+     * Job for loading the page.
      */
-    private var statusSubscription: Subscription? = null
+    private var loadJob: Job? = null
+
+    /**
+     * Job for status changes of the page.
+     */
+    private var statusJob: Job? = null
 
     /**
      * Job for progress changes of the page.
@@ -77,12 +82,17 @@ class PagerPageHolder(
     private var progressJob: Job? = null
 
     /**
-     * Subscription for status changes of the page.
+     * Job for loading the page.
      */
-    private var extraStatusSubscription: Subscription? = null
+    private var extraLoadJob: Job? = null
 
     /**
-     * Subscription for progress changes of the page.
+     * Job for status changes of the page.
+     */
+    private var extraStatusJob: Job? = null
+
+    /**
+     * Job for progress changes of the page.
      */
     private var extraProgressJob: Job? = null
 
@@ -92,14 +102,9 @@ class PagerPageHolder(
      */
     private var readImageHeaderSubscription: Subscription? = null
 
-    // SY -->
-    var status: Page.State = Page.State.QUEUE
-    var extraStatus: Page.State = Page.State.QUEUE
-    // SY <--
-
     init {
         addView(progressIndicator)
-        observeStatus()
+        launchLoadJob()
     }
 
     /**
@@ -110,35 +115,38 @@ class PagerPageHolder(
         super.onDetachedFromWindow()
         cancelProgressJob(1)
         cancelProgressJob(2)
-        unsubscribeStatus(1)
+        cancelLoadJob(1)
-        unsubscribeStatus(2)
+        cancelLoadJob(2)
         unsubscribeReadImageHeader()
     }
 
     /**
-     * Observes the status of the page and notify the changes.
+     * Starts loading the page and processing changes to the page's status.
      *
      * @see processStatus
      */
-    private fun observeStatus() {
+    private fun launchLoadJob() {
-        statusSubscription?.unsubscribe()
+        loadJob?.cancel()
+        statusJob?.cancel()
 
         val loader = page.chapter.pageLoader ?: return
-        statusSubscription = loader.getPage(page)
+        loadJob = scope.launch {
-            .observeOn(AndroidSchedulers.mainThread())
+            loader.loadPage(page)
-            .subscribe {
+        }
-                status = it
+        statusJob = scope.launch {
-                processStatus(it)
+            page.statusFlow.collectLatest { processStatus(it) }
         }
 
+        // SY -->
         val extraPage = extraPage ?: return
         val loader2 = extraPage.chapter.pageLoader ?: return
-        extraStatusSubscription = loader2.getPage(extraPage)
+        extraLoadJob = scope.launch {
-            .observeOn(AndroidSchedulers.mainThread())
+            loader2.loadPage(extraPage)
-            .subscribe {
-                extraStatus = it
-                processStatus2(it)
         }
+        extraStatusJob = scope.launch {
+            extraPage.statusFlow.collectLatest { processStatus2(it) }
+        }
+        // SY <--
     }
 
     private fun launchProgressJob() {
@@ -170,7 +178,7 @@ class PagerPageHolder(
                 setDownloading()
             }
             Page.State.READY -> {
-                if (extraStatus == Page.State.READY || extraPage == null) {
+                if (extraPage?.status == Page.State.READY || extraPage == null) {
                     setImage()
                 }
                 cancelProgressJob(1)
@@ -196,7 +204,7 @@ class PagerPageHolder(
                 setDownloading()
             }
             Page.State.READY -> {
-                if (this.status == Page.State.READY) {
+                if (page.status == Page.State.READY) {
                     setImage()
                 }
                 cancelProgressJob(2)
@@ -209,12 +217,20 @@ class PagerPageHolder(
     }
 
     /**
-     * Unsubscribes from the status subscription.
+     * Cancels loading the page and processing changes to the page's status.
      */
-    private fun unsubscribeStatus(page: Int) {
+    private fun cancelLoadJob(page: Int) {
-        val subscription = if (page == 1) statusSubscription else extraStatusSubscription
+        if (page == 1) {
-        subscription?.unsubscribe()
+            loadJob?.cancel()
-        if (page == 1) statusSubscription = null else extraStatusSubscription = null
+            loadJob = null
+            statusJob?.cancel()
+            statusJob = null
+        } else if (page == 2) {
+            extraLoadJob?.cancel()
+            extraLoadJob = null
+            extraStatusJob?.cancel()
+            extraStatusJob = null
+        }
     }
 
     private fun cancelProgressJob(page: Int) {

app/src/main/java/eu/kanade/tachiyomi/ui/reader/viewer/webtoon/WebtoonPageHolder.kt

@@ -73,9 +73,14 @@ class WebtoonPageHolder(
     private val scope = MainScope()
 
     /**
-     * Subscription for status changes of the page.
+     * Job for loading the page.
      */
-    private var statusSubscription: Subscription? = null
+    private var loadJob: Job? = null
+
+    /**
+     * Job for status changes of the page.
+     */
+    private var statusJob: Job? = null
 
     /**
      * Job for progress changes of the page.
@@ -101,7 +106,7 @@ class WebtoonPageHolder(
      */
     fun bind(page: ReaderPage) {
         this.page = page
-        observeStatus()
+        launchLoadJob()
         refreshLayoutParams()
     }
 
@@ -121,7 +126,7 @@ class WebtoonPageHolder(
      * Called when the view is recycled and added to the view pool.
      */
     override fun recycle() {
-        unsubscribeStatus()
+        cancelLoadJob()
         cancelProgressJob()
         unsubscribeReadImageHeader()
 
@@ -131,20 +136,21 @@ class WebtoonPageHolder(
     }
 
     /**
-     * Observes the status of the page and notify the changes.
+     * Starts loading the page and processing changes to the page's status.
      *
      * @see processStatus
      */
-    private fun observeStatus() {
+    private fun launchLoadJob() {
-        unsubscribeStatus()
+        cancelLoadJob()
 
         val page = page ?: return
         val loader = page.chapter.pageLoader ?: return
-        statusSubscription = loader.getPage(page)
+        loadJob = scope.launch {
-            .observeOn(AndroidSchedulers.mainThread())
+            loader.loadPage(page)
-            .subscribe { processStatus(it) }
+        }
-
+        statusJob = scope.launch {
-        addSubscription(statusSubscription)
+            page.statusFlow.collectLatest { processStatus(it) }
+        }
     }
 
     /**
@@ -185,11 +191,13 @@ class WebtoonPageHolder(
     }
 
     /**
-     * Unsubscribes from the status subscription.
+     * Cancels loading the page and processing changes to the page's status.
      */
-    private fun unsubscribeStatus() {
+    private fun cancelLoadJob() {
-        removeSubscription(statusSubscription)
+        loadJob?.cancel()
-        statusSubscription = null
+        loadJob = null
+        statusJob?.cancel()
+        statusJob = null
     }
 
     /**

source-api/src/main/java/eu/kanade/tachiyomi/source/model/Page.kt

@@ -21,10 +21,14 @@ open class Page(
         get() = index + 1
 
     @Transient
-    @Volatile
+    private val _statusFlow = MutableStateFlow(State.QUEUE)
-    var status: State = State.QUEUE
+
+    @Transient
+    val statusFlow = _statusFlow.asStateFlow()
+    var status: State
+        get() = _statusFlow.value
         set(value) {
-            field = value
+            _statusFlow.value = value
             statusSubject?.onNext(value)
         }