Convert last used source preference to flow

app/src/main/java/eu/kanade/tachiyomi/data/preference/PreferencesHelper.kt

@@ -137,7 +137,7 @@ class PreferencesHelper(val context: Context) {
 
     fun autoUpdateTrack() = prefs.getBoolean(Keys.autoUpdateTrack, true)
 
-    fun lastUsedCatalogueSource() = rxPrefs.getLong(Keys.lastUsedCatalogueSource, -1)
+    fun lastUsedCatalogueSource() = flowPrefs.getLong(Keys.lastUsedCatalogueSource, -1)
 
     fun lastUsedCategory() = flowPrefs.getInt(Keys.lastUsedCategory, 0)
 

app/src/main/java/eu/kanade/tachiyomi/ui/browse/source/SourcePresenter.kt

@@ -7,10 +7,18 @@ import eu.kanade.tachiyomi.source.LocalSource
 import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.ui.base.presenter.BasePresenter
 import java.util.TreeMap
-import java.util.concurrent.TimeUnit
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.drop
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.onStart
 import rx.Observable
 import rx.Subscription
-import rx.android.schedulers.AndroidSchedulers
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 
@@ -26,6 +34,8 @@ class SourcePresenter(
     private val preferences: PreferencesHelper = Injekt.get()
 ) : BasePresenter<SourceController>() {
 
+    private val scope = CoroutineScope(Job() + Dispatchers.Main)
+
     var sources = getEnabledSources()
 
     /**
@@ -79,16 +89,21 @@ class SourcePresenter(
     }
 
     private fun loadLastUsedSource() {
-        val sharedObs = preferences.lastUsedCatalogueSource().asObservable().share()
+        // Immediate initial load
+        preferences.lastUsedCatalogueSource().get().let { updateLastUsedSource(it) }
 
-        // Emit the first item immediately but delay subsequent emissions by 500ms.
+        // Subsequent updates
-        Observable.merge(
+        preferences.lastUsedCatalogueSource().asFlow()
-            sharedObs.take(1),
+            .drop(1)
-            sharedObs.skip(1).delay(500, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
+            .onStart { delay(500) }
-        )
             .distinctUntilChanged()
-            .map { item -> (sourceManager.get(item) as? CatalogueSource)?.let { SourceItem(it) } }
+            .onEach { updateLastUsedSource(it) }
-            .subscribeLatestCache(SourceController::setLastUsedSource)
+            .launchIn(scope)
+    }
+
+    private fun updateLastUsedSource(sourceId: Long) {
+        val source = (sourceManager.get(sourceId) as? CatalogueSource)?.let { SourceItem(it) }
+        source?.let { view?.setLastUsedSource(it) }
     }
 
     fun updateSources() {