Revert "Use flows instead of relays for extensions loading"

This reverts commit 07e76f35fa.

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

@@ -2,6 +2,7 @@ package eu.kanade.tachiyomi.extension
 
 import android.content.Context
 import android.graphics.drawable.Drawable
+import com.jakewharton.rxrelay.BehaviorRelay
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.data.preference.plusAssign
 import eu.kanade.tachiyomi.extension.api.ExtensionGithubApi
@@ -13,11 +14,9 @@ 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.launchIO
 import eu.kanade.tachiyomi.util.lang.launchNow
 import eu.kanade.tachiyomi.util.system.toast
 import kotlinx.coroutines.async
-import kotlinx.coroutines.flow.MutableStateFlow
 import rx.Observable
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
@@ -50,7 +49,7 @@ class ExtensionManager(
     /**
      * Relay used to notify the installed extensions.
      */
-    val installedExtensionsFlow = MutableStateFlow<List<Extension.Installed>>(emptyList())
+    private val installedExtensionsRelay = BehaviorRelay.create<List<Extension.Installed>>()
 
     private val iconMap = mutableMapOf<String, Drawable>()
 
@@ -60,7 +59,7 @@ class ExtensionManager(
     var installedExtensions = emptyList<Extension.Installed>()
         private set(value) {
             field = value
-            launchIO { installedExtensionsFlow.emit(value) }
+            installedExtensionsRelay.call(value)
         }
 
     fun getAppIconForSource(source: Source): Drawable? {
@@ -74,7 +73,7 @@ class ExtensionManager(
     /**
      * Relay used to notify the available extensions.
      */
-    val availableExtensionsFlow = MutableStateFlow<List<Extension.Available>>(emptyList())
+    private val availableExtensionsRelay = BehaviorRelay.create<List<Extension.Available>>()
 
     /**
      * List of the currently available extensions.
@@ -82,14 +81,14 @@ class ExtensionManager(
     var availableExtensions = emptyList<Extension.Available>()
         private set(value) {
             field = value
-            launchIO { availableExtensionsFlow.emit(value) }
+            availableExtensionsRelay.call(value)
             updatedInstalledExtensionsStatuses(value)
         }
 
     /**
      * Relay used to notify the untrusted extensions.
      */
-    val untrustedExtensionsFlow = MutableStateFlow<List<Extension.Untrusted>>(emptyList())
+    private val untrustedExtensionsRelay = BehaviorRelay.create<List<Extension.Untrusted>>()
 
     /**
      * List of the currently untrusted extensions.
@@ -97,7 +96,7 @@ class ExtensionManager(
     var untrustedExtensions = emptyList<Extension.Untrusted>()
         private set(value) {
             field = value
-            launchIO { untrustedExtensionsFlow.emit(value) }
+            untrustedExtensionsRelay.call(value)
         }
 
     /**
@@ -132,6 +131,27 @@ class ExtensionManager(
             .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].
      */

app/src/main/java/eu/kanade/tachiyomi/ui/browse/extension/ExtensionPresenter.kt

@@ -8,17 +8,13 @@ import eu.kanade.tachiyomi.extension.ExtensionManager
 import eu.kanade.tachiyomi.extension.model.Extension
 import eu.kanade.tachiyomi.extension.model.InstallStep
 import eu.kanade.tachiyomi.ui.base.presenter.BasePresenter
-import eu.kanade.tachiyomi.util.lang.launchIO
 import eu.kanade.tachiyomi.util.system.LocaleHelper
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.collectLatest
-import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.debounce
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.withContext
 import rx.Observable
+import rx.Subscription
+import rx.android.schedulers.AndroidSchedulers
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
+import java.util.concurrent.TimeUnit
 
 private typealias ExtensionTuple =
     Triple<List<Extension.Installed>, List<Extension.Untrusted>, List<Extension.Available>>
@@ -39,23 +35,20 @@ open class ExtensionPresenter(
         super.onCreate(savedState)
 
         extensionManager.findAvailableExtensions()
-
+        bindToExtensionsObservable()
-        launchIO {
-            val installedFlow = extensionManager.installedExtensionsFlow
-            val untrustedFlow = extensionManager.untrustedExtensionsFlow
-            val availableFlow = extensionManager.availableExtensionsFlow
-
-            combine(
-                installedFlow,
-                untrustedFlow,
-                availableFlow
-            ) { installed, untrusted, available ->
-                Triple(installed, untrusted, available)
     }
-                .debounce(100)
+
+    private fun bindToExtensionsObservable(): Subscription {
+        val installedObservable = extensionManager.getInstalledExtensionsObservable()
+        val untrustedObservable = extensionManager.getUntrustedExtensionsObservable()
+        val availableObservable = extensionManager.getAvailableExtensionsObservable()
+            .startWith(emptyList<Extension.Available>())
+
+        return Observable.combineLatest(installedObservable, untrustedObservable, availableObservable) { installed, untrusted, available -> Triple(installed, untrusted, available) }
+            .debounce(100, TimeUnit.MILLISECONDS)
             .map(::toItems)
-                .collectLatest { withContext(Dispatchers.Main) { view?.setExtensions(extensions) } }
+            .observeOn(AndroidSchedulers.mainThread())
-        }
+            .subscribeLatestCache({ view, _ -> view.setExtensions(extensions) })
     }
 
     @Synchronized

app/src/main/java/eu/kanade/tachiyomi/ui/browse/extension/details/ExtensionDetailsPresenter.kt

@@ -3,13 +3,7 @@ package eu.kanade.tachiyomi.ui.browse.extension.details
 import android.os.Bundle
 import eu.kanade.tachiyomi.extension.ExtensionManager
 import eu.kanade.tachiyomi.ui.base.presenter.BasePresenter
-import eu.kanade.tachiyomi.util.lang.launchIO
+import rx.android.schedulers.AndroidSchedulers
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.collect
-import kotlinx.coroutines.flow.drop
-import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.take
-import kotlinx.coroutines.withContext
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 
@@ -23,14 +17,19 @@ class ExtensionDetailsPresenter(
     override fun onCreate(savedState: Bundle?) {
         super.onCreate(savedState)
 
-        // Watch for uninstalled event
+        bindToUninstalledExtension()
-        launchIO {
-            extensionManager.installedExtensionsFlow
-                .drop(1)
-                .filter { extensions -> extensions.none { it.pkgName == pkgName } }
-                .take(1)
-                .collect { withContext(Dispatchers.Main) { view?.onExtensionUninstalled() } }
     }
+
+    private fun bindToUninstalledExtension() {
+        extensionManager.getInstalledExtensionsObservable()
+            .skip(1)
+            .filter { extensions -> extensions.none { it.pkgName == pkgName } }
+            .map { }
+            .take(1)
+            .observeOn(AndroidSchedulers.mainThread())
+            .subscribeFirst({ view, _ ->
+                view.onExtensionUninstalled()
+            })
     }
 
     fun uninstallExtension() {