Add data listener to browse source

domain/src/commonMain/kotlin/ca/gosyer/jui/domain/ServerListeners.kt

@@ -13,6 +13,7 @@ import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.filter
@@ -27,9 +28,10 @@ class ServerListeners @Inject constructor() {
 
     private fun <T> Flow<T>.startWith(value: T) = onStart { emit(value) }
 
-    private val mangaListener = MutableSharedFlow<List<Long>>(
+    private val _mangaListener = MutableSharedFlow<List<Long>>(
         extraBufferCapacity = Channel.UNLIMITED
     )
+    val mangaListener = _mangaListener.asSharedFlow()
 
     private val chapterIndexesListener = MutableSharedFlow<Pair<Long, List<Int>?>>(
         extraBufferCapacity = Channel.UNLIMITED
@@ -49,18 +51,18 @@ class ServerListeners @Inject constructor() {
 
     fun <T> combineMangaUpdates(flow: Flow<T>, predate: (suspend (List<Long>) -> Boolean)? = null) =
         if (predate != null) {
-            mangaListener
+            _mangaListener
                 .filter(predate)
                 .startWith(Unit)
         } else {
-            mangaListener.startWith(Unit)
+            _mangaListener.startWith(Unit)
         }
             .buffer(capacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
             .flatMapLatest { flow }
 
     fun updateManga(vararg ids: Long) {
         scope.launch {
-            mangaListener.emit(ids.toList())
+            _mangaListener.emit(ids.toList())
         }
     }
 

domain/src/commonMain/kotlin/ca/gosyer/jui/domain/source/interactor/SourcePager.kt

@@ -0,0 +1,86 @@
+/*
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package ca.gosyer.jui.domain.source.interactor
+
+import ca.gosyer.jui.domain.ServerListeners
+import ca.gosyer.jui.domain.manga.interactor.GetManga
+import ca.gosyer.jui.domain.manga.model.Manga
+import ca.gosyer.jui.domain.source.model.MangaPage
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.runningFold
+import kotlinx.coroutines.flow.stateIn
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import me.tatarka.inject.annotations.Inject
+import org.lighthousegames.logging.logging
+
+class SourcePager @Inject constructor(
+    private val getManga: GetManga,
+    private val serverListeners: ServerListeners,
+    private val fetcher: suspend (page: Int) -> MangaPage?,
+) : CoroutineScope by CoroutineScope(Dispatchers.Default + SupervisorJob()) {
+    private val sourceMutex = Mutex()
+
+    private val _sourceManga = MutableStateFlow<List<Manga>>(emptyList())
+
+    private val mangaIds = _sourceManga.map { mangas -> mangas.map { it.id } }
+        .stateIn(this, SharingStarted.Eagerly, emptyList())
+
+    private val changedManga = serverListeners.mangaListener.runningFold(emptyMap<Long, Manga>()) { manga, updatedMangaIds ->
+        coroutineScope {
+            manga + updatedMangaIds.filter { it in mangaIds.value }.map {
+                async {
+                    getManga.await(it)
+                }
+            }.awaitAll().filterNotNull().associateBy { it.id }
+        }
+    }.stateIn(this, SharingStarted.Eagerly, emptyMap())
+
+    val mangas = combine(_sourceManga, changedManga) { sourceManga, changedManga ->
+        sourceManga.map { changedManga[it.id] ?: it }
+    }.stateIn(this, SharingStarted.Eagerly, emptyList())
+
+    private val _pageNum = MutableStateFlow(0)
+    val pageNum = _pageNum.asStateFlow()
+
+    private val _hasNextPage = MutableStateFlow(true)
+    val hasNextPage = _hasNextPage.asStateFlow()
+
+    private val _loading = MutableStateFlow(true)
+    val loading = _loading.asStateFlow()
+
+    fun loadNextPage() {
+        launch {
+            if (hasNextPage.value && sourceMutex.tryLock()) {
+                _pageNum.value++
+                val page = fetcher(_pageNum.value)
+                if (page != null) {
+                    _sourceManga.value = _sourceManga.value + page.mangaList
+                    _hasNextPage.value = page.hasNextPage
+                } else {
+                    _pageNum.value--
+                }
+                sourceMutex.unlock()
+            }
+            _loading.value = false
+        }
+    }
+
+    companion object {
+        private val log = logging()
+    }
+}

presentation/src/commonMain/kotlin/ca/gosyer/jui/ui/sources/browse/SourceScreenViewModel.kt

@@ -8,10 +8,10 @@ package ca.gosyer.jui.ui.sources.browse
 
 import ca.gosyer.jui.domain.library.model.DisplayMode
 import ca.gosyer.jui.domain.library.service.LibraryPreferences
-import ca.gosyer.jui.domain.manga.model.Manga
 import ca.gosyer.jui.domain.source.interactor.GetLatestManga
 import ca.gosyer.jui.domain.source.interactor.GetPopularManga
 import ca.gosyer.jui.domain.source.interactor.GetSearchManga
+import ca.gosyer.jui.domain.source.interactor.SourcePager
 import ca.gosyer.jui.domain.source.model.MangaPage
 import ca.gosyer.jui.domain.source.model.Source
 import ca.gosyer.jui.domain.source.service.CatalogPreferences
@@ -19,15 +19,14 @@ import ca.gosyer.jui.ui.base.state.SavedStateHandle
 import ca.gosyer.jui.ui.base.state.getStateFlow
 import ca.gosyer.jui.uicore.vm.ContextWrapper
 import ca.gosyer.jui.uicore.vm.ViewModel
-import kotlinx.collections.immutable.ImmutableList
 import kotlinx.collections.immutable.persistentListOf
-import kotlinx.collections.immutable.plus
 import kotlinx.collections.immutable.toImmutableList
-import kotlinx.collections.immutable.toPersistentList
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.stateIn
 import me.tatarka.inject.annotations.Inject
 import org.lighthousegames.logging.logging
 
@@ -38,6 +37,7 @@ class SourceScreenViewModel(
     private val getSearchManga: GetSearchManga,
     private val catalogPreferences: CatalogPreferences,
     private val libraryPreferences: LibraryPreferences,
+    private val getSourcePager: (suspend (page: Int) -> MangaPage?) -> SourcePager,
     contextWrapper: ContextWrapper,
     private val savedStateHandle: SavedStateHandle,
     initialQuery: String?
@@ -49,6 +49,7 @@ class SourceScreenViewModel(
         getSearchManga: GetSearchManga,
         catalogPreferences: CatalogPreferences,
         libraryPreferences: LibraryPreferences,
+        getSourcePager: (suspend (page: Int) -> MangaPage?) -> SourcePager,
         contextWrapper: ContextWrapper,
         savedStateHandle: SavedStateHandle,
         params: Params
@@ -59,6 +60,7 @@ class SourceScreenViewModel(
         getSearchManga,
         catalogPreferences,
         libraryPreferences,
+        getSourcePager,
         contextWrapper,
         savedStateHandle,
         params.initialQuery
@@ -68,15 +70,6 @@ class SourceScreenViewModel(
     val gridColumns = libraryPreferences.gridColumns().stateIn(scope)
     val gridSize = libraryPreferences.gridSize().stateIn(scope)
 
-    private val _mangas = MutableStateFlow<ImmutableList<Manga>>(persistentListOf())
-    val mangas = _mangas.asStateFlow()
-
-    private val _hasNextPage = MutableStateFlow(false)
-    val hasNextPage = _hasNextPage.asStateFlow()
-
-    private val _loading = MutableStateFlow(true)
-    val loading = _loading.asStateFlow()
-
     private val _isLatest by savedStateHandle.getStateFlow { false }
     val isLatest = _isLatest.asStateFlow()
 
@@ -87,71 +80,74 @@ class SourceScreenViewModel(
 
     private val _query = MutableStateFlow(sourceSearchQuery.value)
 
-    private val _pageNum = MutableStateFlow(1)
-    val pageNum = _pageNum.asStateFlow()
+    private val pager = MutableStateFlow(getPager())
+
+    val mangas = pager.flatMapLatest { it.mangas.map { mangas -> mangas.toImmutableList() } }
+        .stateIn(scope, SharingStarted.Eagerly, persistentListOf())
+    val loading = pager.flatMapLatest { it.loading }
+        .stateIn(scope, SharingStarted.Eagerly, true)
+    val hasNextPage = pager.flatMapLatest { it.hasNextPage }
+        .stateIn(scope, SharingStarted.Eagerly, true)
 
-    private val sourceMutex = Mutex()
 
     init {
-        scope.launch {
-            getPage()?.let { (mangas, hasNextPage) ->
-                _mangas.value = mangas.toImmutableList()
-                _hasNextPage.value = hasNextPage
-            }
-
-            _loading.value = false
-        }
+        pager.value.loadNextPage()
     }
 
     fun loadNextPage() {
-        scope.launch {
-            if (hasNextPage.value && sourceMutex.tryLock()) {
-                _pageNum.value++
-                val page = getPage()
-                if (page != null) {
-                    _mangas.value = _mangas.value.toPersistentList() + page.mangaList
-                    _hasNextPage.value = page.hasNextPage
-                } else {
-                    _pageNum.value--
-                }
-                sourceMutex.unlock()
-            }
-            _loading.value = false
-        }
+        pager.value.loadNextPage()
     }
 
     fun setMode(toLatest: Boolean) {
         if (isLatest.value != toLatest) {
             _isLatest.value = toLatest
-            // [loadNextPage] increments by 1
-            _pageNum.value = 0
-            _loading.value = true
             _query.value = null
-            _mangas.value = persistentListOf()
-            loadNextPage()
+            updatePager()
         }
     }
 
-    private suspend fun getPage(): MangaPage? {
-        return when {
-            isLatest.value -> getLatestManga.await(source, pageNum.value, onError = { toast(it.message.orEmpty()) })
-            _query.value != null || _usingFilters.value -> getSearchManga.await(
-                sourceId = source.id,
-                searchTerm = _query.value,
-                page = pageNum.value,
-                onError = { toast(it.message.orEmpty()) }
-            )
-            else -> getPopularManga.await(source.id, pageNum.value, onError = { toast(it.message.orEmpty()) })
+    private fun getPager(): SourcePager {
+        val fetcher: suspend (page: Int) -> MangaPage? = when {
+            _query.value != null || _usingFilters.value -> {
+                { page ->
+                    getSearchManga.await(
+                        sourceId = source.id,
+                        searchTerm = _query.value,
+                        page = page,
+                        onError = { toast(it.message.orEmpty()) }
+                    )
+                }
+            }
+            isLatest.value -> {
+                { page ->
+                    getLatestManga.await(
+                        source,
+                        page,
+                        onError = { toast(it.message.orEmpty()) })
+                }
+            }
+            else -> {
+                { page ->
+                    getPopularManga.await(
+                        source.id,
+                        page,
+                        onError = { toast(it.message.orEmpty()) })
+                }
+            }
         }
+
+        return getSourcePager(fetcher)
+    }
+
+    private fun updatePager() {
+        pager.value.cancel()
+        pager.value = getPager()
+        pager.value.loadNextPage()
     }
 
     fun startSearch(query: String?) {
-        _pageNum.value = 0
-        _hasNextPage.value = true
-        _loading.value = true
         _query.value = query
-        _mangas.value = persistentListOf()
-        loadNextPage()
+        updatePager()
     }
 
     fun setUsingFilters(usingFilters: Boolean) {
@@ -171,6 +167,11 @@ class SourceScreenViewModel(
 
     data class Params(val source: Source, val initialQuery: String?)
 
+    override fun onDispose() {
+        super.onDispose()
+        pager.value.cancel()
+    }
+
     private companion object {
         private val log = logging()
     }