Use flows instead of relays for extensions loading

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

@@ -2,7 +2,6 @@ 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
@@ -14,9 +13,11 @@ 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
@@ -49,7 +50,7 @@ class ExtensionManager(
     /**
      * Relay used to notify the installed extensions.
      */
-    private val installedExtensionsRelay = BehaviorRelay.create<List<Extension.Installed>>()
+    val installedExtensionsFlow = MutableStateFlow<List<Extension.Installed>>(emptyList())
 
     private val iconMap = mutableMapOf<String, Drawable>()
 
@@ -59,7 +60,7 @@ class ExtensionManager(
     var installedExtensions = emptyList<Extension.Installed>()
         private set(value) {
             field = value
-            installedExtensionsRelay.call(value)
+            launchIO { installedExtensionsFlow.emit(value) }
         }
 
     fun getAppIconForSource(source: Source): Drawable? {
@@ -73,7 +74,7 @@ class ExtensionManager(
     /**
      * Relay used to notify the available extensions.
      */
-    private val availableExtensionsRelay = BehaviorRelay.create<List<Extension.Available>>()
+    val availableExtensionsFlow = MutableStateFlow<List<Extension.Available>>(emptyList())
 
     /**
      * List of the currently available extensions.
@@ -81,14 +82,14 @@ class ExtensionManager(
     var availableExtensions = emptyList<Extension.Available>()
         private set(value) {
             field = value
-            availableExtensionsRelay.call(value)
+            launchIO { availableExtensionsFlow.emit(value) }
             updatedInstalledExtensionsStatuses(value)
         }
 
     /**
      * Relay used to notify the untrusted extensions.
      */
-    private val untrustedExtensionsRelay = BehaviorRelay.create<List<Extension.Untrusted>>()
+    val untrustedExtensionsFlow = MutableStateFlow<List<Extension.Untrusted>>(emptyList())
 
     /**
      * List of the currently untrusted extensions.
@@ -96,7 +97,7 @@ class ExtensionManager(
     var untrustedExtensions = emptyList<Extension.Untrusted>()
         private set(value) {
             field = value
-            untrustedExtensionsRelay.call(value)
+            launchIO { untrustedExtensionsFlow.emit(value) }
         }
 
     /**
@@ -131,27 +132,6 @@ 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,13 +8,17 @@ 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>>
@@ -35,20 +39,23 @@ open class ExtensionPresenter(
         super.onCreate(savedState)
 
         extensionManager.findAvailableExtensions()
-        bindToExtensionsObservable()
-    }
 
-    private fun bindToExtensionsObservable(): Subscription {
-        val installedObservable = extensionManager.getInstalledExtensionsObservable()
-        val untrustedObservable = extensionManager.getUntrustedExtensionsObservable()
-        val availableObservable = extensionManager.getAvailableExtensionsObservable()
-            .startWith(emptyList<Extension.Available>())
+        launchIO {
+            val installedFlow = extensionManager.installedExtensionsFlow
+            val untrustedFlow = extensionManager.untrustedExtensionsFlow
+            val availableFlow = extensionManager.availableExtensionsFlow
 
-        return Observable.combineLatest(installedObservable, untrustedObservable, availableObservable) { installed, untrusted, available -> Triple(installed, untrusted, available) }
-            .debounce(100, TimeUnit.MILLISECONDS)
-            .map(::toItems)
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribeLatestCache({ view, _ -> view.setExtensions(extensions) })
+            combine(
+                installedFlow,
+                untrustedFlow,
+                availableFlow
+            ) { installed, untrusted, available ->
+                Triple(installed, untrusted, available)
+            }
+                .debounce(100)
+                .map(::toItems)
+                .collectLatest { withContext(Dispatchers.Main) { view?.setExtensions(extensions) } }
+        }
     }
 
     @Synchronized

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

@@ -3,7 +3,13 @@ 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 rx.android.schedulers.AndroidSchedulers
+import eu.kanade.tachiyomi.util.lang.launchIO
+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
 
@@ -17,19 +23,14 @@ class ExtensionDetailsPresenter(
     override fun onCreate(savedState: Bundle?) {
         super.onCreate(savedState)
 
-        bindToUninstalledExtension()
-    }
-
-    private fun bindToUninstalledExtension() {
-        extensionManager.getInstalledExtensionsObservable()
-            .skip(1)
-            .filter { extensions -> extensions.none { it.pkgName == pkgName } }
-            .map { }
-            .take(1)
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribeFirst({ view, _ ->
-                view.onExtensionUninstalled()
-            })
+        // Watch for uninstalled event
+        launchIO {
+            extensionManager.installedExtensionsFlow
+                .drop(1)
+                .filter { extensions -> extensions.none { it.pkgName == pkgName } }
+                .take(1)
+                .collect { withContext(Dispatchers.Main) { view?.onExtensionUninstalled() } }
+        }
     }
 
     fun uninstallExtension() {