Remove use of RxJava from TrackPresenter

app/src/main/java/eu/kanade/tachiyomi/ui/manga/track/TrackPresenter.kt

@@ -11,12 +11,10 @@ import eu.kanade.tachiyomi.ui.base.presenter.BasePresenter
 import eu.kanade.tachiyomi.util.lang.await
 import eu.kanade.tachiyomi.util.lang.launchIO
 import eu.kanade.tachiyomi.util.lang.launchUI
-import eu.kanade.tachiyomi.util.lang.runAsObservable
 import eu.kanade.tachiyomi.util.system.toast
-import rx.Observable
+import kotlinx.coroutines.Job
-import rx.Subscription
+import kotlinx.coroutines.async
-import rx.android.schedulers.AndroidSchedulers
+import kotlinx.coroutines.awaitAll
-import rx.schedulers.Schedulers
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 
@@ -33,59 +31,57 @@ class TrackPresenter(
 
     private val loggedServices by lazy { trackManager.services.filter { it.isLogged } }
 
-    private var trackSubscription: Subscription? = null
+    private var trackJob: Job? = null
-
+    private var searchJob: Job? = null
-    private var searchSubscription: Subscription? = null
+    private var refreshJob: Job? = null
-
-    private var refreshSubscription: Subscription? = null
 
     override fun onCreate(savedState: Bundle?) {
         super.onCreate(savedState)
         fetchTrackings()
     }
 
-    fun fetchTrackings() {
+    private fun fetchTrackings() {
-        trackSubscription?.let { remove(it) }
+        trackJob?.cancel()
-        trackSubscription = db.getTracks(manga)
+        trackJob = launchIO {
-            .asRxObservable()
+            val tracks = db.getTracks(manga).await()
-            .map { tracks ->
+            trackList = loggedServices.map { service ->
-                loggedServices.map { service ->
                 TrackItem(tracks.find { it.sync_id == service.id }, service)
             }
+            view?.onNextTrackings(trackList)
         }
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnNext { trackList = it }
-            .subscribeLatestCache(TrackController::onNextTrackings)
     }
 
     fun refresh() {
-        refreshSubscription?.let { remove(it) }
+        refreshJob?.cancel()
-        refreshSubscription = Observable.from(trackList)
+        refreshJob = launchIO {
+            try {
+                trackList
                     .filter { it.track != null }
-            .flatMap { item ->
+                    .map {
-                runAsObservable({ item.service.refresh(item.track!!) })
+                        async {
-                    .flatMap { db.insertTrack(it).asRxObservable() }
+                            val track = it.service.refresh(it.track!!)
-                    .map { item }
+                            db.insertTrack(track).await()
-                    .onErrorReturn { item }
+                        }
+                    }
+                    .awaitAll()
+
+                view?.onRefreshDone()
+            } catch (e: Throwable) {
+                view?.onRefreshError(e)
+            }
         }
-            .toList()
-            .subscribeOn(Schedulers.io())
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribeFirst(
-                { view, _ -> view.onRefreshDone() },
-                TrackController::onRefreshError
-            )
     }
 
     fun search(query: String, service: TrackService) {
-        searchSubscription?.let { remove(it) }
+        searchJob?.cancel()
-        searchSubscription = runAsObservable({ service.search(query) })
+        searchJob = launchIO {
-            .subscribeOn(Schedulers.io())
+            try {
-            .observeOn(AndroidSchedulers.mainThread())
+                val results = service.search(query)
-            .subscribeLatestCache(
+                launchUI { view?.onSearchResults(results) }
-                TrackController::onSearchResults,
+            } catch (e: Throwable) {
-                TrackController::onSearchResultsError
+                launchUI { view?.onSearchResultsError(e) }
-            )
+            }
+        }
     }
 
     fun registerTracking(item: Track?, service: TrackService) {
@@ -115,15 +111,13 @@ class TrackPresenter(
                 db.insertTrack(track).await()
                 view?.onRefreshDone()
             } catch (e: Throwable) {
-                launchUI {
+                launchUI { view?.onRefreshError(e) }
-                    view?.onRefreshError(e)
 
                 // Restart on error to set old values
                 fetchTrackings()
             }
         }
     }
-    }
 
     fun setStatus(item: TrackItem, index: Int) {
         val track = item.track!!