Add database subscriptions for feed and better thumbnail fetching

app/src/main/java/eu/kanade/presentation/browse/FeedScreen.kt

@@ -28,6 +28,7 @@ import androidx.compose.material3.RadioButton
 import androidx.compose.material3.Text
 import androidx.compose.material3.TextButton
 import androidx.compose.runtime.Composable
+import androidx.compose.runtime.State
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.mutableStateOf
 import androidx.compose.runtime.remember
@@ -84,6 +85,7 @@ fun FeedScreen(
         else -> {
             FeedList(
                 state = presenter,
+                getMangaState = { item, source -> presenter.getManga(item, source) },
                 onClickAdd = onClickAdd,
                 onClickCreate = onClickCreate,
                 onClickSavedSearch = onClickSavedSearch,
@@ -99,6 +101,7 @@ fun FeedScreen(
 @Composable
 fun FeedList(
     state: FeedState,
+    getMangaState: @Composable ((Manga, CatalogueSource?) -> State<Manga>),
     onClickAdd: (CatalogueSource) -> Unit,
     onClickCreate: (CatalogueSource, SavedSearch?) -> Unit,
     onClickSavedSearch: (SavedSearch, CatalogueSource) -> Unit,
@@ -117,6 +120,7 @@ fun FeedList(
             FeedItem(
                 modifier = Modifier.animateItemPlacement(),
                 item = item,
+                getMangaState = { getMangaState(it, item.source) },
                 onClickSavedSearch = onClickSavedSearch,
                 onClickSource = onClickSource,
                 onClickDelete = onClickDelete,
@@ -165,6 +169,7 @@ fun FeedList(
 fun FeedItem(
     modifier: Modifier,
     item: FeedItemUI,
+    getMangaState: @Composable ((Manga) -> State<Manga>),
     onClickSavedSearch: (SavedSearch, CatalogueSource) -> Unit,
     onClickSource: (CatalogueSource) -> Unit,
     onClickDelete: (FeedSavedSearch) -> Unit,
@@ -223,8 +228,9 @@ fun FeedItem(
                     contentPadding = PaddingValues(horizontal = 12.dp),
                 ) {
                     items(item.results) {
+                        val manga by getMangaState(it)
                         FeedCardItem(
-                            manga = it,
+                            manga = manga,
                             onClickManga = onClickManga,
                         )
                     }

app/src/main/java/eu/kanade/presentation/browse/SourceFeedScreen.kt

@@ -21,6 +21,8 @@ import androidx.compose.material3.Text
 import androidx.compose.material3.TopAppBarScrollBehavior
 import androidx.compose.runtime.Composable
 import androidx.compose.runtime.ReadOnlyComposable
+import androidx.compose.runtime.State
+import androidx.compose.runtime.getValue
 import androidx.compose.ui.Alignment
 import androidx.compose.ui.Modifier
 import androidx.compose.ui.res.painterResource
@@ -126,6 +128,7 @@ fun SourceFeedScreen(
                     SourceFeedList(
                         state = presenter,
                         paddingValues = paddingValues,
+                        getMangaState = { presenter.getManga(it) },
                         onClickBrowse = onClickBrowse,
                         onClickLatest = onClickLatest,
                         onClickSavedSearch = onClickSavedSearch,
@@ -142,6 +145,7 @@ fun SourceFeedScreen(
 fun SourceFeedList(
     state: SourceFeedState,
     paddingValues: PaddingValues,
+    getMangaState: @Composable ((Manga) -> State<Manga>),
     onClickBrowse: () -> Unit,
     onClickLatest: () -> Unit,
     onClickSavedSearch: (SavedSearch) -> Unit,
@@ -158,6 +162,7 @@ fun SourceFeedList(
             SourceFeedItem(
                 modifier = Modifier.animateItemPlacement(),
                 item = item,
+                getMangaState = getMangaState,
                 onClickTitle = when (item) {
                     is SourceFeedUI.Browse -> onClickBrowse
                     is SourceFeedUI.Latest -> onClickLatest
@@ -176,6 +181,7 @@ fun SourceFeedList(
 fun SourceFeedItem(
     modifier: Modifier,
     item: SourceFeedUI,
+    getMangaState: @Composable ((Manga) -> State<Manga>),
     onClickTitle: () -> Unit,
     onClickDelete: (FeedSavedSearch) -> Unit,
     onClickManga: (Manga) -> Unit,
@@ -228,8 +234,9 @@ fun SourceFeedItem(
                     contentPadding = PaddingValues(horizontal = 12.dp),
                 ) {
                     items(results) {
+                        val manga by getMangaState(it)
                         FeedCardItem(
-                            manga = it,
+                            manga = manga,
                             onClickManga = onClickManga,
                         )
                     }

app/src/main/java/eu/kanade/tachiyomi/ui/browse/feed/FeedPresenter.kt

@@ -1,5 +1,8 @@
 package eu.kanade.tachiyomi.ui.browse.feed
 
+import androidx.compose.runtime.Composable
+import androidx.compose.runtime.State
+import androidx.compose.runtime.produceState
 import eu.kanade.domain.manga.interactor.GetManga
 import eu.kanade.domain.manga.interactor.InsertManga
 import eu.kanade.domain.manga.interactor.UpdateManga
@@ -18,22 +21,25 @@ import eu.kanade.tachiyomi.data.database.models.Manga
 import eu.kanade.tachiyomi.data.database.models.toDomainManga
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.source.CatalogueSource
-import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.source.model.FilterList
 import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.source.model.SManga
 import eu.kanade.tachiyomi.util.lang.launchIO
-import eu.kanade.tachiyomi.util.lang.runAsObservable
+import eu.kanade.tachiyomi.util.lang.launchNonCancellableIO
+import eu.kanade.tachiyomi.util.lang.withIOContext
 import eu.kanade.tachiyomi.util.system.LocaleHelper
 import eu.kanade.tachiyomi.util.system.logcat
 import exh.savedsearches.models.FeedSavedSearch
 import exh.savedsearches.models.SavedSearch
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.json.Json
 import logcat.LogPriority
@@ -41,13 +47,13 @@ import rx.Observable
 import rx.Subscription
 import rx.android.schedulers.AndroidSchedulers
 import rx.schedulers.Schedulers
-import rx.subjects.PublishSubject
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 import xyz.nulldev.ts.api.http.serializer.FilterSerializer
+import eu.kanade.domain.manga.model.Manga as DomainManga
 
 /**
- * Presenter of [FeedController]
+ * Presenter of [feedTab]
  * Function calls should be done from here. UI calls should be done from the controller.
  *
  * @param sourceManager manages the different sources.
@@ -74,16 +80,6 @@ open class FeedPresenter(
      */
     private var fetchSourcesSubscription: Subscription? = null
 
-    /**
-     * Subject which fetches image of given manga.
-     */
-    private val fetchImageSubject = PublishSubject.create<Triple<List<Manga>, Source, FeedSavedSearch>>()
-
-    /**
-     * Subscription for fetching images of manga.
-     */
-    private var fetchImageSubscription: Subscription? = null
-
     fun onCreate() {
         getFeedSavedSearchGlobal.subscribe()
             .distinctUntilChanged()
@@ -106,7 +102,6 @@ open class FeedPresenter(
 
     fun onDestroy() {
         fetchSourcesSubscription?.unsubscribe()
-        fetchImageSubscription?.unsubscribe()
     }
 
     fun openAddDialog() {
@@ -144,7 +139,7 @@ open class FeedPresenter(
     }
 
     fun createFeed(source: CatalogueSource, savedSearch: SavedSearch?) {
-        launchIO {
+        presenterScope.launchNonCancellableIO {
             insertFeedSavedSearch.await(
                 FeedSavedSearch(
                     id = -1,
@@ -157,7 +152,7 @@ open class FeedPresenter(
     }
 
     fun deleteFeed(feed: FeedSavedSearch) {
-        launchIO {
+        presenterScope.launchNonCancellableIO {
             deleteFeedSavedSearchById.await(feed.id)
         }
     }
@@ -176,7 +171,7 @@ open class FeedPresenter(
         feed: FeedSavedSearch,
         savedSearch: SavedSearch?,
         source: CatalogueSource?,
-        results: List<eu.kanade.domain.manga.model.Manga>?,
+        results: List<DomainManga>?,
     ): FeedItemUI {
         return FeedItemUI(
             feed,
@@ -196,9 +191,6 @@ open class FeedPresenter(
      * Initiates get manga per feed.
      */
     private fun getFeed(feedSavedSearch: List<FeedItemUI>) {
-        // Create image fetch subscription
-        initializeFetchImageSubscription()
-
         fetchSourcesSubscription?.unsubscribe()
         fetchSourcesSubscription = Observable.from(feedSavedSearch)
             .flatMap(
@@ -215,7 +207,6 @@ open class FeedPresenter(
                             .onErrorReturn { MangasPage(emptyList(), false) } // Ignore timeouts or other exceptions
                             .map { it.mangas } // Get manga from search result.
                             .map { list -> list.map { networkToLocalManga(it, itemUI.source.id) } } // Convert to local manga.
-                            .doOnNext { fetchImage(it, itemUI.source, itemUI.feed) } // Load manga covers.
                             .map { list -> itemUI.copy(results = list.mapNotNull { it.toDomainManga() }) }
                     } else {
                         Observable.just(itemUI.copy(results = emptyList()))
@@ -252,71 +243,18 @@ open class FeedPresenter(
         }.getOrElse { FilterList() }
     }
 
-    /**
+    @Composable
-     * Initialize a list of manga.
+    fun getManga(initialManga: DomainManga, source: CatalogueSource?): State<DomainManga> {
-     *
+        return produceState(initialValue = initialManga) {
-     * @param manga the list of manga to initialize.
+            getManga.subscribe(initialManga.url, initialManga.source)
-     */
+                .collectLatest { manga ->
-    private fun fetchImage(manga: List<Manga>, source: CatalogueSource, feed: FeedSavedSearch) {
+                    if (manga == null) return@collectLatest
-        fetchImageSubject.onNext(Triple(manga, source, feed))
+                    withIOContext {
-    }
+                        initializeManga(source, manga)
-
-    /**
-     * Subscribes to the initializer of manga details and updates the view if needed.
-     */
-    private fun initializeFetchImageSubscription() {
-        fetchImageSubscription?.unsubscribe()
-        fetchImageSubscription = fetchImageSubject.observeOn(Schedulers.io())
-            .flatMap { pair ->
-                val source = pair.second
-                Observable.from(pair.first).filter { it.thumbnail_url == null && !it.initialized }
-                    .map { Pair(it, source) }
-                    .concatMap { getMangaDetailsObservable(it.first, it.second) }
-                    .map { Pair(pair.third, it) }
-            }
-            .onBackpressureBuffer()
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribe(
-                { (feed, manga) ->
-                    synchronized(state) {
-                        state.items = items?.map { itemUI ->
-                            if (feed.id == itemUI.feed.id) {
-                                itemUI.copy(
-                                    results = itemUI.results?.map {
-                                        if (it.id == manga.id) {
-                                            manga.toDomainManga()!!
-                                        } else {
-                                            it
-                                        }
-                                    },
-                                )
-                            } else {
-                                itemUI
-                            }
-                        }
                     }
-                },
+                    value = manga
-                { error ->
+                }
-                    logcat(LogPriority.ERROR, error)
-                },
-            )
-    }
-
-    /**
-     * Returns an observable of manga that initializes the given manga.
-     *
-     * @param manga the manga to initialize.
-     * @return an observable of the manga to initialize
-     */
-    private fun getMangaDetailsObservable(manga: Manga, source: Source): Observable<Manga> {
-        return runAsObservable {
-            val networkManga = source.getMangaDetails(manga.copy())
-            manga.copyFrom(networkManga)
-            manga.initialized = true
-            updateManga.await(manga.toDomainManga()!!.toMangaUpdate())
-            manga
         }
-            .onErrorResumeNext { Observable.just(manga) }
     }
 
     /**
@@ -346,6 +284,31 @@ open class FeedPresenter(
         return localManga?.toDbManga()!!
     }
 
+    /**
+     * Initialize a manga.
+     *
+     * @param manga to initialize.
+     */
+    private suspend fun initializeManga(source: CatalogueSource?, manga: DomainManga) {
+        source ?: return
+        if (manga.thumbnailUrl != null && manga.initialized) return
+        withContext(NonCancellable) {
+            val db = manga.toDbManga()
+            try {
+                val networkManga = source.getMangaDetails(db.copy())
+                db.copyFrom(networkManga)
+                db.initialized = true
+                updateManga.await(
+                    db
+                        .toDomainManga()
+                        ?.toMangaUpdate()!!,
+                )
+            } catch (e: Exception) {
+                logcat(LogPriority.ERROR, e)
+            }
+        }
+    }
+
     sealed class Dialog {
         data class AddFeed(val options: List<CatalogueSource>) : Dialog()
         data class AddFeedSearch(val source: CatalogueSource, val options: List<SavedSearch?>) : Dialog()

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

@@ -1,7 +1,10 @@
 package eu.kanade.tachiyomi.ui.browse.source.feed
 
 import android.os.Bundle
+import androidx.compose.runtime.Composable
+import androidx.compose.runtime.State
 import androidx.compose.runtime.getValue
+import androidx.compose.runtime.produceState
 import eu.kanade.domain.manga.interactor.GetManga
 import eu.kanade.domain.manga.interactor.InsertManga
 import eu.kanade.domain.manga.interactor.UpdateManga
@@ -20,19 +23,21 @@ import eu.kanade.tachiyomi.data.database.models.Manga
 import eu.kanade.tachiyomi.data.database.models.toDomainManga
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.source.CatalogueSource
-import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.model.FilterList
 import eu.kanade.tachiyomi.source.model.MangasPage
 import eu.kanade.tachiyomi.source.model.SManga
 import eu.kanade.tachiyomi.ui.base.presenter.BasePresenter
 import eu.kanade.tachiyomi.util.lang.launchNonCancellableIO
-import eu.kanade.tachiyomi.util.lang.runAsObservable
+import eu.kanade.tachiyomi.util.lang.withIOContext
 import eu.kanade.tachiyomi.util.system.logcat
 import exh.savedsearches.models.FeedSavedSearch
 import exh.savedsearches.models.SavedSearch
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.json.Json
 import logcat.LogPriority
@@ -40,10 +45,10 @@ import rx.Observable
 import rx.Subscription
 import rx.android.schedulers.AndroidSchedulers
 import rx.schedulers.Schedulers
-import rx.subjects.PublishSubject
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 import xyz.nulldev.ts.api.http.serializer.FilterSerializer
+import eu.kanade.domain.manga.model.Manga as DomainManga
 
 /**
  * Presenter of [SourceFeedController]
@@ -74,16 +79,6 @@ open class SourceFeedPresenter(
      */
     private var fetchSourcesSubscription: Subscription? = null
 
-    /**
-     * Subject which fetches image of given manga.
-     */
-    private val fetchImageSubject = PublishSubject.create<Triple<List<Manga>, Source, SourceFeedUI>>()
-
-    /**
-     * Subscription for fetching images of manga.
-     */
-    private var fetchImageSubscription: Subscription? = null
-
     override fun onCreate(savedState: Bundle?) {
         super.onCreate(savedState)
 
@@ -101,7 +96,6 @@ open class SourceFeedPresenter(
 
     override fun onDestroy() {
         fetchSourcesSubscription?.unsubscribe()
-        fetchImageSubscription?.unsubscribe()
         super.onDestroy()
     }
 
@@ -151,9 +145,6 @@ open class SourceFeedPresenter(
      * Initiates get manga per feed.
      */
     private fun getFeed(feedSavedSearch: List<SourceFeedUI>) {
-        // Create image fetch subscription
-        initializeFetchImageSubscription()
-
         fetchSourcesSubscription?.unsubscribe()
         fetchSourcesSubscription = Observable.from(feedSavedSearch)
             .flatMap(
@@ -173,7 +164,6 @@ open class SourceFeedPresenter(
                         .onErrorReturn { MangasPage(emptyList(), false) } // Ignore timeouts or other exceptions
                         .map { it.mangas } // Get manga from search result.
                         .map { list -> list.map { networkToLocalManga(it, source.id) } } // Convert to local manga.
-                        .doOnNext { fetchImage(it, source, sourceFeed) } // Load manga covers.
                         .map { list -> sourceFeed.withResults(list.mapNotNull { it.toDomainManga() }) }
                 },
                 5,
@@ -208,71 +198,18 @@ open class SourceFeedPresenter(
         }.getOrElse { FilterList() }
     }
 
-    /**
+    @Composable
-     * Initialize a list of manga.
+    fun getManga(initialManga: DomainManga): State<DomainManga> {
-     *
+        return produceState(initialValue = initialManga) {
-     * @param manga the list of manga to initialize.
+            getManga.subscribe(initialManga.url, initialManga.source)
-     */
+                .collectLatest { manga ->
-    private fun fetchImage(manga: List<Manga>, source: Source, sourceFeed: SourceFeedUI) {
+                    if (manga == null) return@collectLatest
-        fetchImageSubject.onNext(Triple(manga, source, sourceFeed))
+                    withIOContext {
-    }
+                        initializeManga(manga)
-
-    /**
-     * Subscribes to the initializer of manga details and updates the view if needed.
-     */
-    private fun initializeFetchImageSubscription() {
-        fetchImageSubscription?.unsubscribe()
-        fetchImageSubscription = fetchImageSubject.observeOn(Schedulers.io())
-            .flatMap { pair ->
-                val source = pair.second
-                Observable.from(pair.first).filter { it.thumbnail_url == null && !it.initialized }
-                    .map { Pair(it, source) }
-                    .concatMap { getMangaDetailsObservable(it.first, it.second) }
-                    .map { Pair(pair.third, it) }
-            }
-            .onBackpressureBuffer()
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribe(
-                { (sourceFeed, manga) ->
-                    synchronized(state) {
-                        state.items = items?.map { itemUI ->
-                            if (sourceFeed.id == itemUI.id) {
-                                itemUI.withResults(
-                                    results = itemUI.results?.map {
-                                        if (it.id == manga.id) {
-                                            manga.toDomainManga()!!
-                                        } else {
-                                            it
-                                        }
-                                    },
-                                )
-                            } else {
-                                itemUI
-                            }
-                        }
                     }
-                },
+                    value = manga
-                { error ->
+                }
-                    logcat(LogPriority.ERROR, error)
-                },
-            )
-    }
-
-    /**
-     * Returns an observable of manga that initializes the given manga.
-     *
-     * @param manga the manga to initialize.
-     * @return an observable of the manga to initialize
-     */
-    private fun getMangaDetailsObservable(manga: Manga, source: Source): Observable<Manga> {
-        return runAsObservable {
-            val networkManga = source.getMangaDetails(manga.copy())
-            manga.copyFrom(networkManga)
-            manga.initialized = true
-            updateManga.await(manga.toDomainManga()!!.toMangaUpdate())
-            manga
         }
-            .onErrorResumeNext { Observable.just(manga) }
     }
 
     /**
@@ -302,6 +239,30 @@ open class SourceFeedPresenter(
         return localManga?.toDbManga()!!
     }
 
+    /**
+     * Initialize a manga.
+     *
+     * @param manga to initialize.
+     */
+    private suspend fun initializeManga(manga: DomainManga) {
+        if (manga.thumbnailUrl != null && manga.initialized) return
+        withContext(NonCancellable) {
+            val db = manga.toDbManga()
+            try {
+                val networkManga = source.getMangaDetails(db.copy())
+                db.copyFrom(networkManga)
+                db.initialized = true
+                updateManga.await(
+                    db
+                        .toDomainManga()
+                        ?.toMangaUpdate()!!,
+                )
+            } catch (e: Exception) {
+                logcat(LogPriority.ERROR, e)
+            }
+        }
+    }
+
     suspend fun loadSearch(searchId: Long) =
         getExhSavedSearch.awaitOne(searchId, source::getFilterList)