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

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

@@ -8,6 +8,7 @@ import eu.kanade.tachiyomi.ui.reader.model.ReaderChapter
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import eu.kanade.tachiyomi.util.lang.awaitSingle
 import eu.kanade.tachiyomi.util.lang.launchIO
+import eu.kanade.tachiyomi.util.lang.withIOContext
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -16,10 +17,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
@@ -53,7 +51,7 @@ class HttpPageLoader(
             }
                 .filter { it.status == Page.State.QUEUE }
                 .collect {
-                    loadPage(it)
+                    _loadPage(it)
                 }
         }
     }
@@ -103,11 +101,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
@@ -120,17 +117,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)
@@ -138,8 +132,7 @@ class HttpPageLoader(
                     }
                 }
             }
-            .subscribeOn(Schedulers.io())
+        }
-            .unsubscribeOn(Schedulers.io())
     }
 
     /**
@@ -197,7 +190,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

@@ -60,9 +60,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,7 +82,7 @@ class PagerPageHolder(
 
     init {
         addView(progressIndicator)
-        observeStatus()
+        launchLoadJob()
     }
 
     /**
@@ -87,22 +92,26 @@ class PagerPageHolder(
     override fun onDetachedFromWindow() {
         super.onDetachedFromWindow()
         cancelProgressJob()
-        unsubscribeStatus()
+        cancelLoadJob()
         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 { processStatus(it) }
+        }
+        statusJob = scope.launch {
+            page.statusFlow.collectLatest { processStatus(it) }
+        }
     }
 
     private fun launchProgressJob() {
@@ -137,11 +146,13 @@ class PagerPageHolder(
     }
 
     /**
-     * Unsubscribes from the status subscription.
+     * Cancels loading the page and processing changes to the page's status.
      */
-    private fun unsubscribeStatus() {
+    private fun cancelLoadJob() {
-        statusSubscription?.unsubscribe()
+        loadJob?.cancel()
-        statusSubscription = null
+        loadJob = null
+        statusJob?.cancel()
+        statusJob = null
     }
 
     private fun cancelProgressJob() {

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

@@ -20,10 +20,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)
         }