Remove usage of RxJava from backup/restore

app/src/main/java/eu/kanade/tachiyomi/data/backup/AbstractBackupManager.kt

@@ -12,8 +12,6 @@ import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.source.model.toSChapter
 import eu.kanade.tachiyomi.util.chapter.syncChaptersWithSource
-import eu.kanade.tachiyomi.util.lang.runAsObservable
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 
 abstract class AbstractBackupManager(protected val context: Context) {
@@ -34,25 +32,22 @@ abstract class AbstractBackupManager(protected val context: Context) {
         databaseHelper.getManga(manga.url, manga.source).executeAsBlocking()
 
     /**
-     * [Observable] that fetches chapter information
+     * Fetches chapter information.
      *
      * @param source source of manga
      * @param manga manga that needs updating
      * @param chapters list of chapters in the backup
-     * @return [Observable] that contains manga
+     * @return Updated manga chapters.
      */
-    internal fun restoreChapterFetchObservable(source: Source, manga: Manga, chapters: List<Chapter>): Observable<Pair<List<Chapter>, List<Chapter>>> {
+    internal suspend fun restoreChapters(source: Source, manga: Manga, chapters: List<Chapter>): Pair<List<Chapter>, List<Chapter>> {
-        return runAsObservable({
+        val fetchedChapters = source.getChapterList(manga.toMangaInfo())
-            source.getChapterList(manga.toMangaInfo())
             .map { it.toSChapter() }
-        })
+        val syncedChapters = syncChaptersWithSource(databaseHelper, fetchedChapters, manga, source)
-            .map { syncChaptersWithSource(databaseHelper, it, manga, source) }
+        if (syncedChapters.first.isNotEmpty()) {
-            .doOnNext { (first) ->
-                if (first.isNotEmpty()) {
             chapters.forEach { it.manga_id = manga.id }
             updateChapters(chapters)
         }
-            }
+        return syncedChapters
     }
 
     /**

app/src/main/java/eu/kanade/tachiyomi/data/backup/AbstractBackupRestore.kt

@@ -10,9 +10,8 @@ import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.data.track.TrackManager
 import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.util.chapter.NoChaptersException
-import eu.kanade.tachiyomi.util.lang.runAsObservable
+import eu.kanade.tachiyomi.util.lang.await
 import kotlinx.coroutines.Job
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 import java.io.File
 import java.text.SimpleDateFormat
@@ -59,20 +58,21 @@ abstract class AbstractBackupRestore<T : AbstractBackupManager>(protected val co
     }
 
     /**
-     * [Observable] that fetches chapter information
+     * Fetches chapter information.
      *
      * @param source source of manga
      * @param manga manga that needs updating
-     * @return [Observable] that contains manga
+     * @return Updated manga chapters.
      */
-    internal fun chapterFetchObservable(source: Source, manga: Manga, chapters: List<Chapter>): Observable<Pair<List<Chapter>, List<Chapter>>> {
+    internal suspend fun updateChapters(source: Source, manga: Manga, chapters: List<Chapter>): Pair<List<Chapter>, List<Chapter>> {
-        return backupManager.restoreChapterFetchObservable(source, manga, chapters)
+        return try {
+            backupManager.restoreChapters(source, manga, chapters)
+        } catch (e: Exception) {
             // If there's any error, return empty update and continue.
-            .onErrorReturn {
+            val errorMessage = if (e is NoChaptersException) {
-                val errorMessage = if (it is NoChaptersException) {
                 context.getString(R.string.no_chapters_error)
             } else {
-                    it.message
+                e.message
             }
             errors.add(Date() to "${manga.title} - $errorMessage")
             Pair(emptyList(), emptyList())
@@ -80,25 +80,23 @@ abstract class AbstractBackupRestore<T : AbstractBackupManager>(protected val co
     }
 
     /**
-     * [Observable] that refreshes tracking information
+     * Refreshes tracking information.
+     *
      * @param manga manga that needs updating.
      * @param tracks list containing tracks from restore file.
-     * @return [Observable] that contains updated track item
      */
-    internal fun trackingFetchObservable(manga: Manga, tracks: List<Track>): Observable<Track> {
+    internal suspend fun updateTracking(manga: Manga, tracks: List<Track>) {
-        return Observable.from(tracks)
+        tracks.forEach { track ->
-            .flatMap { track ->
             val service = trackManager.getService(track.sync_id)
             if (service != null && service.isLogged) {
-                    runAsObservable({ service.refresh(track) })
+                try {
-                        .doOnNext { db.insertTrack(it).executeAsBlocking() }
+                    val updatedTrack = service.refresh(track)
-                        .onErrorReturn {
+                    db.insertTrack(updatedTrack).await()
-                            errors.add(Date() to "${manga.title} - ${it.message}")
+                } catch (e: Exception) {
-                            track
+                    errors.add(Date() to "${manga.title} - ${e.message}")
                 }
             } else {
                 errors.add(Date() to "${manga.title} - ${context.getString(R.string.tracker_not_logged_in, service?.name)}")
-                    Observable.empty()
             }
         }
     }

app/src/main/java/eu/kanade/tachiyomi/data/backup/full/FullBackupManager.kt

@@ -29,13 +29,11 @@ import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.data.database.models.toMangaInfo
 import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.model.toSManga
-import eu.kanade.tachiyomi.util.lang.runAsObservable
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.protobuf.ProtoBuf
 import okio.buffer
 import okio.gzip
 import okio.sink
-import rx.Observable
 import timber.log.Timber
 import kotlin.math.max
 
@@ -185,28 +183,25 @@ class FullBackupManager(context: Context) : AbstractBackupManager(context) {
     }
 
     /**
-     * [Observable] that fetches manga information
+     * Fetches manga information
      *
      * @param source source of manga
      * @param manga manga that needs updating
-     * @return [Observable] that contains manga
+     * @return Updated manga info.
      */
-    fun restoreMangaFetchObservable(source: Source?, manga: Manga, online: Boolean): Observable<Manga> {
+    suspend fun restoreMangaFetch(source: Source?, manga: Manga, online: Boolean): Manga {
         return if (online && source != null) {
-            return runAsObservable({
             val networkManga = source.getMangaDetails(manga.toMangaInfo())
-                manga.copyFrom(networkManga.toSManga())
+            manga.also {
-                manga.favorite = manga.favorite
+                it.copyFrom(networkManga.toSManga())
-                manga.initialized = true
+                it.favorite = manga.favorite
-                manga.id = insertManga(manga)
+                it.initialized = true
-                manga
+                it.id = insertManga(manga)
-            })
+            }
         } else {
-            Observable.just(manga)
+            manga.also {
-                .map {
                 it.initialized = it.description != null
                 it.id = insertManga(it)
-                    it
             }
         }
     }

app/src/main/java/eu/kanade/tachiyomi/data/backup/full/FullBackupRestore.kt

@@ -13,11 +13,11 @@ import eu.kanade.tachiyomi.data.database.models.Chapter
 import eu.kanade.tachiyomi.data.database.models.Manga
 import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.source.Source
+import eu.kanade.tachiyomi.util.lang.launchIO
 import kotlinx.serialization.ExperimentalSerializationApi
 import okio.buffer
 import okio.gzip
 import okio.source
-import rx.Observable
 import java.util.Date
 
 @OptIn(ExperimentalSerializationApi::class)
@@ -120,7 +120,7 @@ class FullBackupRestore(context: Context, notifier: BackupNotifier, private val
     }
 
     /**
-     * [Observable] that fetches manga information
+     * Fetches manga information
      *
      * @param manga manga that needs updating
      * @param chapters chapters of manga that needs updating
@@ -136,28 +136,24 @@ class FullBackupRestore(context: Context, notifier: BackupNotifier, private val
         backupCategories: List<BackupCategory>,
         online: Boolean
     ) {
-        backupManager.restoreMangaFetchObservable(source, manga, online)
+        launchIO {
-            .doOnError {
+            try {
-                errors.add(Date() to "${manga.title} - ${it.message}")
+                val fetchedManga = backupManager.restoreMangaFetch(source, manga, online)
-            }
+                fetchedManga.id ?: (return@launchIO)
-            .filter { it.id != null }
+
-            .flatMap {
                 if (online && source != null) {
-                    chapterFetchObservable(source, it, chapters)
+                    updateChapters(source, fetchedManga, chapters)
-                        // Convert to the manga that contains new chapters.
-                        .map { manga }
                 } else {
-                    backupManager.restoreChaptersForMangaOffline(it, chapters)
+                    backupManager.restoreChaptersForMangaOffline(fetchedManga, chapters)
-                    Observable.just(manga)
+                }
+
+                restoreExtraForManga(fetchedManga, categories, history, tracks, backupCategories)
+
+                updateTracking(fetchedManga, tracks)
+            } catch (e: Exception) {
+                errors.add(Date() to "${manga.title} - ${e.message}")
             }
         }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks, backupCategories)
-            }
-            .flatMap {
-                trackingFetchObservable(it, tracks)
-            }
-            .subscribe()
     }
 
     private fun restoreMangaNoFetch(
@@ -170,27 +166,19 @@ class FullBackupRestore(context: Context, notifier: BackupNotifier, private val
         backupCategories: List<BackupCategory>,
         online: Boolean
     ) {
-        Observable.just(backupManga)
+        launchIO {
-            .flatMap { manga ->
             if (online && source != null) {
-                    if (!backupManager.restoreChaptersForManga(manga, chapters)) {
+                if (!backupManager.restoreChaptersForManga(backupManga, chapters)) {
-                        chapterFetchObservable(source, manga, chapters)
+                    updateChapters(source, backupManga, chapters)
-                            .map { manga }
-                    } else {
-                        Observable.just(manga)
                 }
             } else {
-                    backupManager.restoreChaptersForMangaOffline(manga, chapters)
+                backupManager.restoreChaptersForMangaOffline(backupManga, chapters)
-                    Observable.just(manga)
             }
+
+            restoreExtraForManga(backupManga, categories, history, tracks, backupCategories)
+
+            updateTracking(backupManga, tracks)
         }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks, backupCategories)
-            }
-            .flatMap { manga ->
-                trackingFetchObservable(manga, tracks)
-            }
-            .subscribe()
     }
 
     private fun restoreExtraForManga(manga: Manga, categories: List<Int>, history: List<BackupHistory>, tracks: List<Track>, backupCategories: List<BackupCategory>) {

app/src/main/java/eu/kanade/tachiyomi/data/backup/legacy/LegacyBackupManager.kt

@@ -48,8 +48,6 @@ import eu.kanade.tachiyomi.data.database.models.toMangaInfo
 import eu.kanade.tachiyomi.source.LocalSource
 import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.model.toSManga
-import eu.kanade.tachiyomi.util.lang.runAsObservable
-import rx.Observable
 import timber.log.Timber
 import kotlin.math.max
 
@@ -252,21 +250,20 @@ class LegacyBackupManager(context: Context, version: Int = CURRENT_VERSION) : Ab
     }
 
     /**
-     * [Observable] that fetches manga information
+     * Fetches manga information
      *
      * @param source source of manga
      * @param manga manga that needs updating
-     * @return [Observable] that contains manga
+     * @return Updated manga.
      */
-    fun restoreMangaFetchObservable(source: Source, manga: Manga): Observable<Manga> {
+    suspend fun fetchManga(source: Source, manga: Manga): Manga {
-        return runAsObservable({
         val networkManga = source.getMangaDetails(manga.toMangaInfo())
-            manga.copyFrom(networkManga.toSManga())
+        return manga.also {
-            manga.favorite = true
+            it.copyFrom(networkManga.toSManga())
-            manga.initialized = true
+            it.favorite = true
-            manga.id = insertManga(manga)
+            it.initialized = true
-            manga
+            it.id = insertManga(manga)
-        })
+        }
     }
 
     /**

app/src/main/java/eu/kanade/tachiyomi/data/backup/legacy/LegacyBackupRestore.kt

@@ -21,7 +21,7 @@ import eu.kanade.tachiyomi.data.database.models.MangaImpl
 import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.data.database.models.TrackImpl
 import eu.kanade.tachiyomi.source.Source
-import rx.Observable
+import eu.kanade.tachiyomi.util.lang.launchIO
 import java.util.Date
 
 class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : AbstractBackupRestore<LegacyBackupManager>(context, notifier) {
@@ -137,7 +137,7 @@ class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : Abstract
     }
 
     /**
-     * [Observable] that fetches manga information
+     * Fetches manga information.
      *
      * @param manga manga that needs updating
      * @param chapters chapters of manga that needs updating
@@ -151,24 +151,20 @@ class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : Abstract
         history: List<DHistory>,
         tracks: List<Track>
     ) {
-        backupManager.restoreMangaFetchObservable(source, manga)
+        launchIO {
-            .onErrorReturn {
+            try {
-                errors.add(Date() to "${manga.title} - ${it.message}")
+                val fetchedManga = backupManager.fetchManga(source, manga)
-                manga
+                fetchedManga.id ?: (return@launchIO)
+
+                updateChapters(source, fetchedManga, chapters)
+
+                restoreExtraForManga(fetchedManga, categories, history, tracks)
+
+                updateTracking(fetchedManga, tracks)
+            } catch (e: Exception) {
+                errors.add(Date() to "${manga.title} - ${e.message}")
             }
-            .filter { it.id != null }
-            .flatMap {
-                chapterFetchObservable(source, it, chapters)
-                    // Convert to the manga that contains new chapters.
-                    .map { manga }
         }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks)
-            }
-            .flatMap {
-                trackingFetchObservable(it, tracks)
-            }
-            .subscribe()
     }
 
     private fun restoreMangaNoFetch(
@@ -179,22 +175,15 @@ class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : Abstract
         history: List<DHistory>,
         tracks: List<Track>
     ) {
-        Observable.just(backupManga)
+        launchIO {
-            .flatMap { manga ->
+            if (!backupManager.restoreChaptersForManga(backupManga, chapters)) {
-                if (!backupManager.restoreChaptersForManga(manga, chapters)) {
+                updateChapters(source, backupManga, chapters)
-                    chapterFetchObservable(source, manga, chapters)
-                        .map { manga }
-                } else {
-                    Observable.just(manga)
             }
+
+            restoreExtraForManga(backupManga, categories, history, tracks)
+
+            updateTracking(backupManga, tracks)
         }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks)
-            }
-            .flatMap { manga ->
-                trackingFetchObservable(manga, tracks)
-            }
-            .subscribe()
     }
 
     private fun restoreExtraForManga(manga: Manga, categories: List<String>, history: List<DHistory>, tracks: List<Track>) {

app/src/test/java/eu/kanade/tachiyomi/data/backup/BackupTest.kt

@@ -211,7 +211,7 @@ class BackupTest {
         networkManga.description = "This is a description"
         `when`(source.fetchMangaDetails(jsonManga)).thenReturn(Observable.just(networkManga))
 
-        val obs = legacyBackupManager.restoreMangaFetchObservable(source, jsonManga)
+        val obs = legacyBackupManager.fetchManga(source, jsonManga)
         val testSubscriber = TestSubscriber<Manga>()
         obs.subscribe(testSubscriber)
 
@@ -255,7 +255,7 @@ class BackupTest {
         `when`(source.fetchChapterList(manga)).thenReturn(Observable.just(chaptersRemote))
 
         // Call restoreChapterFetchObservable
-        val obs = legacyBackupManager.restoreChapterFetchObservable(source, manga, restoredChapters)
+        val obs = legacyBackupManager.restoreChapters(source, manga, restoredChapters)
         val testSubscriber = TestSubscriber<Pair<List<Chapter>, List<Chapter>>>()
         obs.subscribe(testSubscriber)