Use flows instead of relays for extensions loading

(cherry picked from commit 07e76f35fa2a69a4cda0668f540aa1a4e2eb2b98)

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

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

@@ -4,7 +4,6 @@ import android.content.Context
 import android.graphics.drawable.Drawable
 import androidx.core.content.ContextCompat
 import com.elvishew.xlog.XLog
-import com.jakewharton.rxrelay.BehaviorRelay
 import eu.kanade.tachiyomi.R
 import eu.kanade.tachiyomi.data.preference.PreferencesHelper
 import eu.kanade.tachiyomi.data.preference.plusAssign
@@ -17,6 +16,7 @@ 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 exh.EH_SOURCE_ID
@@ -24,6 +24,7 @@ import exh.EXH_SOURCE_ID
 import exh.MERGED_SOURCE_ID
 import exh.source.BlacklistedSources
 import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.MutableStateFlow
 import rx.Observable
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
@@ -56,7 +57,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>()
 
@@ -66,7 +67,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? {
@@ -88,7 +89,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.
@@ -96,7 +97,7 @@ class ExtensionManager(
     var availableExtensions = emptyList<Extension.Available>()
         private set(value) {
             field = value
-            availableExtensionsRelay.call(value)
+            launchIO { availableExtensionsFlow.emit(value) }
             updatedInstalledExtensionsStatuses(value)
         }
 
@@ -107,7 +108,7 @@ class ExtensionManager(
     /**
      * 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.
@@ -115,7 +116,7 @@ class ExtensionManager(
     var untrustedExtensions = emptyList<Extension.Untrusted>()
         private set(value) {
             field = value
-            untrustedExtensionsRelay.call(value)
+            launchIO { untrustedExtensionsFlow.emit(value) }
         }
 
     /**
@@ -169,27 +170,6 @@ class ExtensionManager(
     }
     // EXH <--
 
-    /**
-     * 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 {
+        launchIO {
-        val installedObservable = extensionManager.getInstalledExtensionsObservable()
+            val installedFlow = extensionManager.installedExtensionsFlow
-        val untrustedObservable = extensionManager.getUntrustedExtensionsObservable()
+            val untrustedFlow = extensionManager.untrustedExtensionsFlow
-        val availableObservable = extensionManager.getAvailableExtensionsObservable()
+            val availableFlow = extensionManager.availableExtensionsFlow
-            .startWith(emptyList<Extension.Available>())
 
-        return Observable.combineLatest(installedObservable, untrustedObservable, availableObservable) { installed, untrusted, available -> Triple(installed, untrusted, available) }
+            combine(
-            .debounce(100, TimeUnit.MILLISECONDS)
+                installedFlow,
-            .map(::toItems)
+                untrustedFlow,
-            .observeOn(AndroidSchedulers.mainThread())
+                availableFlow
-            .subscribeLatestCache({ view, _ -> view.setExtensions(extensions) })
+            ) { 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()
+        // Watch for uninstalled event
-    }
+        launchIO {
-
+            extensionManager.installedExtensionsFlow
-    private fun bindToUninstalledExtension() {
+                .drop(1)
-        extensionManager.getInstalledExtensionsObservable()
+                .filter { extensions -> extensions.none { it.pkgName == pkgName } }
-            .skip(1)
+                .take(1)
-            .filter { extensions -> extensions.none { it.pkgName == pkgName } }
+                .collect { withContext(Dispatchers.Main) { view?.onExtensionUninstalled() } }
-            .map { }
+        }
-            .take(1)
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribeFirst({ view, _ ->
-                view.onExtensionUninstalled()
-            })
     }
 
     fun uninstallExtension() {