Use Flow in ExtensionManager and SourceManager (#7547)

- Replace ExtensionManager relay and observable with Flow
- Inverse SourceManager dependency
    - SourceManager observers ExtensionManager flow
- Separate SourceData from SourceRepository as it created a circular dependency

app/src/main/java/eu/kanade/data/source/SourceDataRepositoryImpl.kt

@@ -0,0 +1,23 @@
+package eu.kanade.data.source
+
+import eu.kanade.data.DatabaseHandler
+import eu.kanade.domain.source.model.SourceData
+import eu.kanade.domain.source.repository.SourceDataRepository
+import kotlinx.coroutines.flow.Flow
+
+class SourceDataRepositoryImpl(
+    private val handler: DatabaseHandler,
+) : SourceDataRepository {
+
+    override fun subscribeAll(): Flow<List<SourceData>> {
+        return handler.subscribeToList { sourcesQueries.findAll(sourceDataMapper) }
+    }
+
+    override suspend fun getSourceData(id: Long): SourceData? {
+        return handler.awaitOneOrNull { sourcesQueries.findOne(id, sourceDataMapper) }
+    }
+
+    override suspend fun upsertSourceData(id: Long, lang: String, name: String) {
+        handler.await { sourcesQueries.upsert(id, lang, name) }
+    }
+}

app/src/main/java/eu/kanade/data/source/SourceRepositoryImpl.kt

@@ -2,7 +2,6 @@ package eu.kanade.data.source
 
 import eu.kanade.data.DatabaseHandler
 import eu.kanade.domain.source.model.Source
-import eu.kanade.domain.source.model.SourceData
 import eu.kanade.domain.source.repository.SourceRepository
 import eu.kanade.tachiyomi.source.LocalSource
 import eu.kanade.tachiyomi.source.SourceManager
@@ -50,12 +49,4 @@ class SourceRepositoryImpl(
             }
         }
     }
-
-    override suspend fun getSourceData(id: Long): SourceData? {
-        return handler.awaitOneOrNull { sourcesQueries.getSourceData(id, sourceDataMapper) }
-    }
-
-    override suspend fun upsertSourceData(id: Long, lang: String, name: String) {
-        handler.await { sourcesQueries.upsert(id, lang, name) }
-    }
 }

app/src/main/java/eu/kanade/domain/DomainModule.kt

@@ -4,6 +4,7 @@ import eu.kanade.data.category.CategoryRepositoryImpl
 import eu.kanade.data.chapter.ChapterRepositoryImpl
 import eu.kanade.data.history.HistoryRepositoryImpl
 import eu.kanade.data.manga.MangaRepositoryImpl
+import eu.kanade.data.source.SourceDataRepositoryImpl
 import eu.kanade.data.source.SourceRepositoryImpl
 import eu.kanade.data.track.TrackRepositoryImpl
 import eu.kanade.domain.category.interactor.CreateCategoryWithName
@@ -47,14 +48,13 @@ import eu.kanade.domain.manga.interactor.UpdateManga
 import eu.kanade.domain.manga.repository.MangaRepository
 import eu.kanade.domain.source.interactor.GetEnabledSources
 import eu.kanade.domain.source.interactor.GetLanguagesWithSources
-import eu.kanade.domain.source.interactor.GetSourceData
 import eu.kanade.domain.source.interactor.GetSourcesWithFavoriteCount
 import eu.kanade.domain.source.interactor.GetSourcesWithNonLibraryManga
 import eu.kanade.domain.source.interactor.SetMigrateSorting
 import eu.kanade.domain.source.interactor.ToggleLanguage
 import eu.kanade.domain.source.interactor.ToggleSource
 import eu.kanade.domain.source.interactor.ToggleSourcePin
-import eu.kanade.domain.source.interactor.UpsertSourceData
+import eu.kanade.domain.source.repository.SourceDataRepository
 import eu.kanade.domain.source.repository.SourceRepository
 import eu.kanade.domain.track.interactor.DeleteTrack
 import eu.kanade.domain.track.interactor.GetTracks
@@ -120,15 +120,14 @@ class DomainModule : InjektModule {
         addFactory { GetExtensionLanguages(get(), get()) }
 
         addSingletonFactory<SourceRepository> { SourceRepositoryImpl(get(), get()) }
+        addSingletonFactory<SourceDataRepository> { SourceDataRepositoryImpl(get()) }
         addFactory { GetEnabledSources(get(), get()) }
         addFactory { GetLanguagesWithSources(get(), get()) }
-        addFactory { GetSourceData(get()) }
         addFactory { GetSourcesWithFavoriteCount(get(), get()) }
         addFactory { GetSourcesWithNonLibraryManga(get()) }
         addFactory { SetMigrateSorting(get()) }
         addFactory { ToggleLanguage(get()) }
         addFactory { ToggleSource(get()) }
         addFactory { ToggleSourcePin(get()) }
-        addFactory { UpsertSourceData(get()) }
     }
 }

app/src/main/java/eu/kanade/domain/extension/interactor/GetExtensionLanguages.kt

@@ -14,7 +14,7 @@ class GetExtensionLanguages(
     fun subscribe(): Flow<List<String>> {
         return combine(
             preferences.enabledLanguages().asFlow(),
-            extensionManager.getAvailableExtensionsObservable().asFlow(),
+            extensionManager.getAvailableExtensionsFlow(),
         ) { enabledLanguage, availableExtensions ->
             availableExtensions
                 .map { it.lang }

app/src/main/java/eu/kanade/domain/extension/interactor/GetExtensionUpdates.kt

@@ -1,6 +1,5 @@
 package eu.kanade.domain.extension.interactor
 
-import eu.kanade.core.util.asFlow
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.extension.ExtensionManager
 import eu.kanade.tachiyomi.extension.model.Extension
@@ -15,7 +14,7 @@ class GetExtensionUpdates(
     fun subscribe(): Flow<List<Extension.Installed>> {
         val showNsfwSources = preferences.showNsfwSource().get()
 
-        return extensionManager.getInstalledExtensionsObservable().asFlow()
+        return extensionManager.getInstalledExtensionsFlow()
             .map { installed ->
                 installed
                     .filter { it.hasUpdate && (showNsfwSources || it.isNsfw.not()) }

app/src/main/java/eu/kanade/domain/extension/interactor/GetExtensions.kt

@@ -19,9 +19,9 @@ class GetExtensions(
 
         return combine(
             preferences.enabledLanguages().asFlow(),
-            extensionManager.getInstalledExtensionsObservable().asFlow(),
+            extensionManager.getInstalledExtensionsFlow(),
-            extensionManager.getUntrustedExtensionsObservable().asFlow(),
+            extensionManager.getUntrustedExtensionsFlow(),
-            extensionManager.getAvailableExtensionsObservable().asFlow(),
+            extensionManager.getAvailableExtensionsFlow(),
         ) { _activeLanguages, _installed, _untrusted, _available ->
 
             val installed = _installed

app/src/main/java/eu/kanade/domain/source/interactor/GetSourceData.kt

@@ -1,20 +0,0 @@
-package eu.kanade.domain.source.interactor
-
-import eu.kanade.domain.source.model.SourceData
-import eu.kanade.domain.source.repository.SourceRepository
-import eu.kanade.tachiyomi.util.system.logcat
-import logcat.LogPriority
-
-class GetSourceData(
-    private val repository: SourceRepository,
-) {
-
-    suspend fun await(id: Long): SourceData? {
-        return try {
-            repository.getSourceData(id)
-        } catch (e: Exception) {
-            logcat(LogPriority.ERROR, e)
-            null
-        }
-    }
-}

app/src/main/java/eu/kanade/domain/source/interactor/UpsertSourceData.kt

@@ -1,19 +0,0 @@
-package eu.kanade.domain.source.interactor
-
-import eu.kanade.domain.source.model.SourceData
-import eu.kanade.domain.source.repository.SourceRepository
-import eu.kanade.tachiyomi.util.system.logcat
-import logcat.LogPriority
-
-class UpsertSourceData(
-    private val repository: SourceRepository,
-) {
-
-    suspend fun await(sourceData: SourceData) {
-        try {
-            repository.upsertSourceData(sourceData.id, sourceData.lang, sourceData.name)
-        } catch (e: Exception) {
-            logcat(LogPriority.ERROR, e)
-        }
-    }
-}

app/src/main/java/eu/kanade/domain/source/repository/SourceDataRepository.kt

@@ -0,0 +1,12 @@
+package eu.kanade.domain.source.repository
+
+import eu.kanade.domain.source.model.SourceData
+import kotlinx.coroutines.flow.Flow
+
+interface SourceDataRepository {
+    fun subscribeAll(): Flow<List<SourceData>>
+
+    suspend fun getSourceData(id: Long): SourceData?
+
+    suspend fun upsertSourceData(id: Long, lang: String, name: String)
+}

app/src/main/java/eu/kanade/domain/source/repository/SourceRepository.kt

@@ -1,7 +1,6 @@
 package eu.kanade.domain.source.repository
 
 import eu.kanade.domain.source.model.Source
-import eu.kanade.domain.source.model.SourceData
 import kotlinx.coroutines.flow.Flow
 import eu.kanade.tachiyomi.source.Source as LoadedSource
 
@@ -14,8 +13,4 @@ interface SourceRepository {
     fun getSourcesWithFavoriteCount(): Flow<List<Pair<Source, Long>>>
 
     fun getSourcesWithNonLibraryManga(): Flow<List<Pair<LoadedSource, Long>>>
-
-    suspend fun getSourceData(id: Long): SourceData?
-
-    suspend fun upsertSourceData(id: Long, lang: String, name: String)
 }

app/src/main/java/eu/kanade/tachiyomi/AppModule.kt

@@ -87,10 +87,10 @@ class AppModule(val app: Application) : InjektModule {
 
         addSingletonFactory { NetworkHelper(app) }
 
-        addSingletonFactory { SourceManager(app).also { get<ExtensionManager>().init(it) } }
-
         addSingletonFactory { ExtensionManager(app) }
 
+        addSingletonFactory { SourceManager(app, get(), get()) }
+
         addSingletonFactory { DownloadManager(app) }
 
         addSingletonFactory { TrackManager(app) }

app/src/main/java/eu/kanade/tachiyomi/extension/ExtensionManager.kt

@@ -14,7 +14,6 @@ import eu.kanade.tachiyomi.extension.util.ExtensionInstallReceiver
 import eu.kanade.tachiyomi.extension.util.ExtensionInstaller
 import eu.kanade.tachiyomi.extension.util.ExtensionLoader
 import eu.kanade.tachiyomi.source.Source
-import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.util.lang.launchNow
 import eu.kanade.tachiyomi.util.preference.plusAssign
 import eu.kanade.tachiyomi.util.system.logcat
@@ -88,22 +87,23 @@ class ExtensionManager(
         return null
     }
 
-    /**
-     * Relay used to notify the available extensions.
-     */
-    private val availableExtensionsRelay = BehaviorRelay.create<List<Extension.Available>>()
-
     /**
      * List of the currently available extensions.
      */
     var availableExtensions = emptyList<Extension.Available>()
         private set(value) {
             field = value
-            availableExtensionsRelay.call(value)
+            availableExtensionsFlow.value = field
             updatedInstalledExtensionsStatuses(value)
             setupAvailableExtensionsSourcesDataMap(value)
         }
 
+    private val availableExtensionsFlow = MutableStateFlow(availableExtensions)
+
+    fun getAvailableExtensionsFlow(): StateFlow<List<Extension.Available>> {
+        return availableExtensionsFlow.asStateFlow()
+    }
+
     private var availableExtensionsSourcesData: Map<Long, SourceData> = mapOf()
 
     private fun setupAvailableExtensionsSourcesDataMap(extensions: List<Extension.Available>) {
@@ -115,30 +115,22 @@ class ExtensionManager(
 
     fun getSourceData(id: Long) = availableExtensionsSourcesData[id]
 
-    /**
-     * Relay used to notify the untrusted extensions.
-     */
-    private val untrustedExtensionsRelay = BehaviorRelay.create<List<Extension.Untrusted>>()
-
     /**
      * List of the currently untrusted extensions.
      */
     var untrustedExtensions = emptyList<Extension.Untrusted>()
         private set(value) {
             field = value
-            untrustedExtensionsRelay.call(value)
+            untrustedExtensionsFlow.value = field
         }
 
-    /**
+    private val untrustedExtensionsFlow = MutableStateFlow(untrustedExtensions)
-     * The source manager where the sources of the extensions are added.
-     */
-    private lateinit var sourceManager: SourceManager
 
-    /**
+    fun getUntrustedExtensionsFlow(): StateFlow<List<Extension.Untrusted>> {
-     * Initializes this manager with the given source manager.
+        return untrustedExtensionsFlow.asStateFlow()
-     */
+    }
-    fun init(sourceManager: SourceManager) {
+
-        this.sourceManager = sourceManager
+    init {
         initExtensions()
         ExtensionInstallReceiver(InstallationListener()).register(context)
     }
@@ -152,36 +144,12 @@ class ExtensionManager(
         installedExtensions = extensions
             .filterIsInstance<LoadResult.Success>()
             .map { it.extension }
-        installedExtensions
-            .flatMap { it.sources }
-            .forEach { sourceManager.registerSource(it) }
 
         untrustedExtensions = extensions
             .filterIsInstance<LoadResult.Untrusted>()
             .map { it.extension }
     }
 
-    /**
-     * Returns the relay of the installed extensions as an observable.
-     */
-    fun getInstalledExtensionsObservable(): Observable<List<Extension.Installed>> {
-        return installedExtensionsRelay.asObservable()
-    }
-
-    /**
-     * Returns the relay of the available extensions as an observable.
-     */
-    fun getAvailableExtensionsObservable(): Observable<List<Extension.Available>> {
-        return availableExtensionsRelay.asObservable()
-    }
-
-    /**
-     * Returns the relay of the untrusted extensions as an observable.
-     */
-    fun getUntrustedExtensionsObservable(): Observable<List<Extension.Untrusted>> {
-        return untrustedExtensionsRelay.asObservable()
-    }
-
     /**
      * Finds the available extensions in the [api] and updates [availableExtensions].
      */
@@ -324,7 +292,6 @@ class ExtensionManager(
      */
     private fun registerNewExtension(extension: Extension.Installed) {
         installedExtensions += extension
-        extension.sources.forEach { sourceManager.registerSource(it) }
     }
 
     /**
@@ -338,11 +305,9 @@ class ExtensionManager(
         val oldExtension = mutInstalledExtensions.find { it.pkgName == extension.pkgName }
         if (oldExtension != null) {
             mutInstalledExtensions -= oldExtension
-            extension.sources.forEach { sourceManager.unregisterSource(it) }
         }
         mutInstalledExtensions += extension
         installedExtensions = mutInstalledExtensions
-        extension.sources.forEach { sourceManager.registerSource(it) }
     }
 
     /**
@@ -355,7 +320,6 @@ class ExtensionManager(
         val installedExtension = installedExtensions.find { it.pkgName == pkgName }
         if (installedExtension != null) {
             installedExtensions -= installedExtension
-            installedExtension.sources.forEach { sourceManager.unregisterSource(it) }
         }
         val untrustedExtension = untrustedExtensions.find { it.pkgName == pkgName }
         if (untrustedExtension != null) {

app/src/main/java/eu/kanade/tachiyomi/source/SourceManager.kt

@@ -1,42 +1,72 @@
 package eu.kanade.tachiyomi.source
 
 import android.content.Context
-import eu.kanade.domain.source.interactor.GetSourceData
-import eu.kanade.domain.source.interactor.UpsertSourceData
 import eu.kanade.domain.source.model.SourceData
+import eu.kanade.domain.source.repository.SourceDataRepository
 import eu.kanade.tachiyomi.R
 import eu.kanade.tachiyomi.extension.ExtensionManager
 import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.source.model.SChapter
 import eu.kanade.tachiyomi.source.model.SManga
 import eu.kanade.tachiyomi.source.online.HttpSource
-import eu.kanade.tachiyomi.util.lang.launchIO
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import rx.Observable
 import tachiyomi.source.model.ChapterInfo
 import tachiyomi.source.model.MangaInfo
-import uy.kohesive.injekt.injectLazy
 
-class SourceManager(private val context: Context) {
+class SourceManager(
+    private val context: Context,
+    private val extensionManager: ExtensionManager,
+    private val sourceRepository: SourceDataRepository,
+) {
 
-    private val extensionManager: ExtensionManager by injectLazy()
+    private val scope = CoroutineScope(Job() + Dispatchers.IO)
-    private val getSourceData: GetSourceData by injectLazy()
+
-    private val upsertSourceData: UpsertSourceData by injectLazy()
+    private var sourcesMap = emptyMap<Long, Source>()
+        set(value) {
+            field = value
+            sourcesMapFlow.value = field
+        }
+
+    private val sourcesMapFlow = MutableStateFlow(sourcesMap)
 
-    private val sourcesMap = mutableMapOf<Long, Source>()
     private val stubSourcesMap = mutableMapOf<Long, StubSource>()
 
-    private val _catalogueSources: MutableStateFlow<List<CatalogueSource>> = MutableStateFlow(listOf())
+    val catalogueSources: Flow<List<CatalogueSource>> = sourcesMapFlow.map { it.values.filterIsInstance<CatalogueSource>() }
-    val catalogueSources: Flow<List<CatalogueSource>> = _catalogueSources
+    val onlineSources: Flow<List<HttpSource>> = catalogueSources.map { sources -> sources.filterIsInstance<HttpSource>() }
-    val onlineSources: Flow<List<HttpSource>> =
-        _catalogueSources.map { sources -> sources.filterIsInstance<HttpSource>() }
 
     init {
-        createInternalSources().forEach { registerSource(it) }
+        scope.launch {
+            extensionManager.getInstalledExtensionsFlow()
+                .collectLatest { extensions ->
+                    val mutableMap = mutableMapOf<Long, Source>(LocalSource.ID to LocalSource(context))
+                    extensions.forEach { extension ->
+                        extension.sources.forEach {
+                            mutableMap[it.id] = it
+                            registerStubSource(it.toSourceData())
+                        }
+                    }
+                    sourcesMap = mutableMap
+                }
+        }
+
+        scope.launch {
+            sourceRepository.subscribeAll()
+                .collectLatest { sources ->
+                    val mutableMap = stubSourcesMap.toMutableMap()
+                    sources.forEach {
+                        mutableMap[it.id] = StubSource(it)
+                    }
+                }
+        }
     }
 
     fun get(sourceKey: Long): Source? {
@@ -58,44 +88,15 @@ class SourceManager(private val context: Context) {
         return stubSourcesMap.values.filterNot { it.id in onlineSourceIds }
     }
 
-    internal fun registerSource(source: Source) {
-        if (!sourcesMap.containsKey(source.id)) {
-            sourcesMap[source.id] = source
-        }
-        registerStubSource(source.toSourceData())
-        triggerCatalogueSources()
-    }
-
     private fun registerStubSource(sourceData: SourceData) {
-        launchIO {
+        scope.launch {
-            val dbSourceData = getSourceData.await(sourceData.id)
+            val (id, lang, name) = sourceData
-
+            sourceRepository.upsertSourceData(id, lang, name)
-            if (dbSourceData != sourceData) {
-                upsertSourceData.await(sourceData)
-            }
-            if (stubSourcesMap[sourceData.id]?.toSourceData() != sourceData) {
-                stubSourcesMap[sourceData.id] = StubSource(sourceData)
         }
     }
-    }
-
-    internal fun unregisterSource(source: Source) {
-        sourcesMap.remove(source.id)
-        triggerCatalogueSources()
-    }
-
-    private fun triggerCatalogueSources() {
-        _catalogueSources.update {
-            sourcesMap.values.filterIsInstance<CatalogueSource>()
-        }
-    }
-
-    private fun createInternalSources(): List<Source> = listOf(
-        LocalSource(context),
-    )
 
     private suspend fun createStubSource(id: Long): StubSource {
-        getSourceData.await(id)?.let {
+        sourceRepository.getSourceData(id)?.let {
             return StubSource(it)
         }
         extensionManager.getSourceData(id)?.let {

app/src/main/sqldelight/data/sources.sq

@@ -4,7 +4,11 @@ CREATE TABLE sources(
     name TEXT NOT NULL
 );
 
-getSourceData:
+findAll:
+SELECT *
+FROM sources;
+
+findOne:
 SELECT *
 FROM sources
 WHERE _id = :id;