Remove usage of RxJava from backup/restore

(cherry picked from commit 990fb22d3eb5e640b81ff4465166a1bfb1f11442)

# Conflicts:
#	app/src/main/java/eu/kanade/tachiyomi/data/backup/AbstractBackupManager.kt
#	app/src/main/java/eu/kanade/tachiyomi/data/backup/AbstractBackupRestore.kt
#	app/src/main/java/eu/kanade/tachiyomi/data/backup/full/FullBackupManager.kt
#	app/src/main/java/eu/kanade/tachiyomi/data/backup/full/FullBackupRestore.kt
#	app/src/main/java/eu/kanade/tachiyomi/data/backup/legacy/LegacyBackupManager.kt
#	app/src/main/java/eu/kanade/tachiyomi/data/backup/legacy/LegacyBackupRestore.kt

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

@@ -13,9 +13,7 @@ import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.source.model.toSChapter
 import eu.kanade.tachiyomi.source.online.all.EHentai
 import eu.kanade.tachiyomi.util.chapter.syncChaptersWithSource
-import eu.kanade.tachiyomi.util.lang.runAsObservable
 import exh.eh.EHentaiThrottleManager
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 
 abstract class AbstractBackupManager(protected val context: Context) {
@@ -36,31 +34,29 @@ 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 open fun restoreChapterFetchObservable(source: Source, manga: Manga, chapters: List<Chapter>, throttleManager: EHentaiThrottleManager): Observable<Pair<List<Chapter>, List<Chapter>>> {
-        return runAsObservable({
-            if (source is EHentai) {
-                source.getChapterList(manga.toMangaInfo(), throttleManager::throttle)
-                    .map { it.toSChapter() }
-            } else {
-                source.getChapterList(manga.toMangaInfo())
-                    .map { it.toSChapter() }
-            }
-        }).map {
-            syncChaptersWithSource(databaseHelper, it, manga, source)
+    internal open suspend fun restoreChapters(source: Source, manga: Manga, chapters: List<Chapter> /* SY --> */, throttleManager: EHentaiThrottleManager /* SY <-- */): Pair<List<Chapter>, List<Chapter>> {
+        // SY -->
+        val fetchedChapters = if (source is EHentai) {
+            source.getChapterList(manga.toMangaInfo(), throttleManager::throttle)
+                .map { it.toSChapter() }
+        } else {
+            source.getChapterList(manga.toMangaInfo())
+                .map { it.toSChapter() }
         }
-            .doOnNext { (first) ->
-                if (first.isNotEmpty()) {
-                    chapters.forEach { it.manga_id = manga.id }
-                    updateChapters(chapters)
-                }
-            }
+        // SY <--
+        val syncedChapters = syncChaptersWithSource(databaseHelper, fetchedChapters, manga, source)
+        if (syncedChapters.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,10 +10,9 @@ 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 exh.eh.EHentaiThrottleManager
 import kotlinx.coroutines.Job
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 import java.io.File
 import java.text.SimpleDateFormat
@@ -64,48 +63,47 @@ 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>>> {
-        return backupManager.restoreChapterFetchObservable(source, manga, chapters /* SY --> */, throttleManager /* SY <-- */)
+    internal suspend fun updateChapters(source: Source, manga: Manga, chapters: List<Chapter>): Pair<List<Chapter>, List<Chapter>> {
+        return try {
+            backupManager.restoreChapters(source, manga, chapters /* SY --> */, throttleManager /* SY <-- */)
+        } catch (e: Exception) {
             // If there's any error, return empty update and continue.
-            .onErrorReturn {
-                val errorMessage = if (it is NoChaptersException) {
-                    context.getString(R.string.no_chapters_error)
-                } else {
-                    it.message
-                }
-                errors.add(Date() to "${manga.title} - $errorMessage")
-                Pair(emptyList(), emptyList())
+            val errorMessage = if (e is NoChaptersException) {
+                context.getString(R.string.no_chapters_error)
+            } else {
+                e.message
             }
+            errors.add(Date() to "${manga.title} - $errorMessage")
+            Pair(emptyList(), emptyList())
+        }
     }
 
     /**
-     * [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> {
-        return Observable.from(tracks)
-            .flatMap { track ->
-                val service = trackManager.getService(track.sync_id)
-                if (service != null && service.isLogged) {
-                    runAsObservable({ service.refresh(track) })
-                        .doOnNext { db.insertTrack(it).executeAsBlocking() }
-                        .onErrorReturn {
-                            errors.add(Date() to "${manga.title} - ${it.message}")
-                            track
-                        }
-                } else {
-                    errors.add(Date() to "${manga.title} - ${context.getString(R.string.tracker_not_logged_in, service?.name)}")
-                    Observable.empty()
+    internal suspend fun updateTracking(manga: Manga, tracks: List<Track>) {
+        tracks.forEach { track ->
+            val service = trackManager.getService(track.sync_id)
+            if (service != null && service.isLogged) {
+                try {
+                    val updatedTrack = service.refresh(track)
+                    db.insertTrack(updatedTrack).await()
+                } catch (e: Exception) {
+                    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)}")
             }
+        }
     }
 
     /**

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

@@ -34,7 +34,6 @@ import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.model.toSManga
 import eu.kanade.tachiyomi.source.online.MetadataSource
 import eu.kanade.tachiyomi.source.online.all.MergedSource
-import eu.kanade.tachiyomi.util.lang.runAsObservable
 import exh.MERGED_SOURCE_ID
 import exh.metadata.metadata.base.getFlatMetadataForManga
 import exh.metadata.metadata.base.insertFlatMetadata
@@ -48,7 +47,6 @@ 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
 
@@ -238,29 +236,26 @@ 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 /* SY --> */ && source !is MergedSource /* SY <-- */) {
-            runAsObservable({
-                val networkManga = source.getMangaDetails(manga.toMangaInfo())
-                manga.copyFrom(networkManga.toSManga())
-                manga.favorite = manga.favorite
-                manga.initialized = true
-                manga.id = insertManga(manga)
-                manga
-            })
+            val networkManga = source.getMangaDetails(manga.toMangaInfo())
+            manga.also {
+                it.copyFrom(networkManga.toSManga())
+                it.favorite = manga.favorite
+                it.initialized = true
+                it.id = insertManga(manga)
+            }
         } else {
-            Observable.just(manga)
-                .map {
-                    it.initialized = it.description != null
-                    it.id = insertManga(it)
-                    it
-                }
+            manga.also {
+                it.initialized = it.description != null
+                it.id = insertManga(it)
+            }
         }
     }
 

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

@@ -17,13 +17,13 @@ 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.source.online.all.MergedSource
+import eu.kanade.tachiyomi.util.lang.launchIO
 import exh.EXHMigrations
 import exh.MERGED_SOURCE_ID
 import kotlinx.serialization.ExperimentalSerializationApi
 import okio.buffer
 import okio.gzip
 import okio.source
-import rx.Observable
 import java.util.Date
 
 @OptIn(ExperimentalSerializationApi::class)
@@ -154,7 +154,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
@@ -172,34 +172,28 @@ class FullBackupRestore(context: Context, notifier: BackupNotifier, private val
         flatMetadata: BackupFlatMetadata?,
         online: Boolean
     ) {
-        backupManager.restoreMangaFetchObservable(source, manga, online)
-            .doOnError {
-                errors.add(Date() to "${manga.title} - ${it.message}")
-            }
-            .filter { it.id != null }
-            .flatMap {
+        launchIO {
+            try {
+                val fetchedManga = backupManager.restoreMangaFetch(source, manga, online)
+                fetchedManga.id ?: (return@launchIO)
+
                 if (online && source != null) {
                     // SY -->
                     if (source !is MergedSource) {
-                        chapterFetchObservable(source, it, chapters)
-                            // Convert to the manga that contains new chapters.
-                            .map { manga }
-                    } else {
-                        Observable.just(manga)
+                        updateChapters(source, fetchedManga, chapters)
                     }
                     // SY <--
                 } else {
-                    backupManager.restoreChaptersForMangaOffline(it, chapters)
-                    Observable.just(manga)
+                    backupManager.restoreChaptersForMangaOffline(fetchedManga, chapters)
                 }
+
+                restoreExtraForManga(fetchedManga, categories, history, tracks, backupCategories, mergedMangaReferences, flatMetadata)
+
+                updateTracking(fetchedManga, tracks)
+            } catch (e: Exception) {
+                errors.add(Date() to "${manga.title} - ${e.message}")
             }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks, backupCategories, mergedMangaReferences, flatMetadata)
-            }
-            .flatMap {
-                trackingFetchObservable(it, tracks)
-            }
-            .subscribe()
+        }
     }
 
     private fun restoreMangaNoFetch(
@@ -214,27 +208,19 @@ class FullBackupRestore(context: Context, notifier: BackupNotifier, private val
         flatMetadata: BackupFlatMetadata?,
         online: Boolean
     ) {
-        Observable.just(backupManga)
-            .flatMap { manga ->
-                if (online && source != null) {
-                    if (/* SY --> */ source !is MergedSource && /* SY <-- */ !backupManager.restoreChaptersForManga(manga, chapters)) {
-                        chapterFetchObservable(source, manga, chapters)
-                            .map { manga }
-                    } else {
-                        Observable.just(manga)
-                    }
-                } else {
-                    backupManager.restoreChaptersForMangaOffline(manga, chapters)
-                    Observable.just(manga)
+        launchIO {
+            if (online && source != null) {
+                if (/* SY --> */ source !is MergedSource && /* SY <-- */ !backupManager.restoreChaptersForManga(backupManga, chapters)) {
+                    updateChapters(source, backupManga, chapters)
                 }
+            } else {
+                backupManager.restoreChaptersForMangaOffline(backupManga, chapters)
             }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks, backupCategories, mergedMangaReferences, flatMetadata)
-            }
-            .flatMap { manga ->
-                trackingFetchObservable(manga, tracks)
-            }
-            .subscribe()
+
+            restoreExtraForManga(backupManga, categories, history, tracks, backupCategories, mergedMangaReferences, flatMetadata)
+
+            updateTracking(backupManga, tracks)
+        }
     }
 
     private fun restoreExtraForManga(manga: Manga, categories: List<Int>, history: List<BackupHistory>, tracks: List<Track>, backupCategories: List<BackupCategory>, mergedMangaReferences: List<BackupMergedMangaReference>, flatMetadata: BackupFlatMetadata?) {

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

@@ -54,16 +54,13 @@ import eu.kanade.tachiyomi.source.LocalSource
 import eu.kanade.tachiyomi.source.Source
 import eu.kanade.tachiyomi.source.model.toSManga
 import eu.kanade.tachiyomi.source.online.all.MergedSource
-import eu.kanade.tachiyomi.util.lang.runAsObservable
 import exh.MERGED_SOURCE_ID
 import exh.eh.EHentaiThrottleManager
 import exh.merged.sql.models.MergedMangaReference
 import exh.savedsearches.JsonSavedSearch
-import kotlinx.coroutines.flow.onEach
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
-import rx.Observable
 import timber.log.Timber
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
@@ -291,21 +288,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> {
-        return runAsObservable({
-            val networkManga = source.getMangaDetails(manga.toMangaInfo())
-            manga.copyFrom(networkManga.toSManga())
-            manga.favorite = true
-            manga.initialized = true
-            manga.id = insertManga(manga)
-            manga
-        })
+    suspend fun fetchManga(source: Source, manga: Manga): Manga {
+        val networkManga = source.getMangaDetails(manga.toMangaInfo())
+        return manga.also {
+            it.copyFrom(networkManga.toSManga())
+            it.favorite = true
+            it.initialized = true
+            it.id = insertManga(manga)
+        }
     }
 
     /**
@@ -315,19 +311,17 @@ class LegacyBackupManager(context: Context, version: Int = CURRENT_VERSION) : Ab
      * @param manga manga that needs updating
      * @return [Observable] that contains manga
      */
-    override fun restoreChapterFetchObservable(source: Source, manga: Manga, chapters: List<Chapter>, throttleManager: EHentaiThrottleManager): Observable<Pair<List<Chapter>, List<Chapter>>> {
+    override suspend fun restoreChapters(source: Source, manga: Manga, chapters: List<Chapter>, throttleManager: EHentaiThrottleManager): Pair<List<Chapter>, List<Chapter>> {
         // SY -->
         return if (source is MergedSource) {
-            runAsObservable({
-                val syncedChapters = source.fetchChaptersAndSync(manga, false)
-                syncedChapters.first.onEach {
-                    it.manga_id = manga.id
-                }
-                updateChapters(syncedChapters.first)
-                syncedChapters
-            })
+            val syncedChapters = source.fetchChaptersAndSync(manga, false)
+            syncedChapters.first.onEach {
+                it.manga_id = manga.id
+            }
+            updateChapters(syncedChapters.first)
+            syncedChapters
         } else {
-            super.restoreChapterFetchObservable(source, manga, chapters, throttleManager)
+            super.restoreChapters(source, manga, chapters, throttleManager)
         }
     }
 

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

@@ -21,8 +21,8 @@ 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 eu.kanade.tachiyomi.util.lang.launchIO
 import exh.EXHMigrations
-import rx.Observable
 import java.util.Date
 
 class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : AbstractBackupRestore<LegacyBackupManager>(context, notifier) {
@@ -169,7 +169,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
@@ -183,24 +183,20 @@ class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : Abstract
         history: List<DHistory>,
         tracks: List<Track>
     ) {
-        backupManager.restoreMangaFetchObservable(source, manga)
-            .onErrorReturn {
-                errors.add(Date() to "${manga.title} - ${it.message}")
-                manga
+        launchIO {
+            try {
+                val fetchedManga = backupManager.fetchManga(source, 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(
@@ -211,22 +207,15 @@ class LegacyBackupRestore(context: Context, notifier: BackupNotifier) : Abstract
         history: List<DHistory>,
         tracks: List<Track>
     ) {
-        Observable.just(backupManga)
-            .flatMap { manga ->
-                if (!backupManager.restoreChaptersForManga(manga, chapters)) {
-                    chapterFetchObservable(source, manga, chapters)
-                        .map { manga }
-                } else {
-                    Observable.just(manga)
-                }
+        launchIO {
+            if (!backupManager.restoreChaptersForManga(backupManga, chapters)) {
+                updateChapters(source, backupManga, chapters)
             }
-            .doOnNext {
-                restoreExtraForManga(it, categories, history, tracks)
-            }
-            .flatMap { manga ->
-                trackingFetchObservable(manga, tracks)
-            }
-            .subscribe()
+
+            restoreExtraForManga(backupManga, categories, history, tracks)
+
+            updateTracking(backupManga, tracks)
+        }
     }
 
     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)