Add presenter subscriptions to the subscription list when using custom subscribe methods

app/src/main/java/eu/kanade/tachiyomi/ui/backup/BackupFragment.kt

@@ -32,7 +32,7 @@ class BackupFragment : BaseRxFragment<BackupPresenter>() {
         return inflater.inflate(R.layout.fragment_backup, container, false)
     }
 
-    override fun onViewCreated(view: View?, savedState: Bundle?) {
+    override fun onViewCreated(view: View, savedState: Bundle?) {
         backup_button.setOnClickListener {
             val today = SimpleDateFormat("yyyy-MM-dd").format(Date())
             val file = File(activity.externalCacheDir, "tachiyomi-$today.json")
@@ -55,12 +55,14 @@ class BackupFragment : BaseRxFragment<BackupPresenter>() {
 
     /**
      * Called from the presenter when the backup is completed.
+     *
+     * @param file the file where the backup is saved.
      */
-    fun onBackupCompleted() {
+    fun onBackupCompleted(file: File) {
         dismissBackupDialog()
         val intent = Intent(Intent.ACTION_SEND)
         intent.type = "text/plain"
-        intent.putExtra(Intent.EXTRA_STREAM, Uri.parse("file://" + presenter.backupFile))
+        intent.putExtra(Intent.EXTRA_STREAM, Uri.parse("file://" + file))
         startActivity(Intent.createChooser(intent, ""))
     }
 

app/src/main/java/eu/kanade/tachiyomi/ui/backup/BackupPresenter.kt

@@ -5,6 +5,7 @@ import eu.kanade.tachiyomi.data.backup.BackupManager
 import eu.kanade.tachiyomi.data.database.DatabaseHelper
 import eu.kanade.tachiyomi.ui.base.presenter.BasePresenter
 import rx.Observable
+import rx.Subscription
 import rx.android.schedulers.AndroidSchedulers
 import rx.schedulers.Schedulers
 import java.io.File
@@ -27,39 +28,18 @@ class BackupPresenter : BasePresenter<BackupFragment>() {
     private lateinit var backupManager: BackupManager
 
     /**
-     * File where the backup is saved.
+     * Subscription where the backup is restored.
      */
-    var backupFile: File? = null
+    private var restoreSubscription: Subscription? = null
-        private set
 
     /**
-     * Stream to restore a backup.
+     * Subscription where the backup is created.
      */
-    private var restoreStream: InputStream? = null
+    private var backupSubscription: Subscription? = null
-
-    /**
-     * Id of the restartable that creates a backup.
-     */
-    private val CREATE_BACKUP = 1
-
-    /**
-     * Id of the restartable that restores a backup.
-     */
-    private val RESTORE_BACKUP = 2
 
     override fun onCreate(savedState: Bundle?) {
         super.onCreate(savedState)
         backupManager = BackupManager(db)
-
-        startableFirst(CREATE_BACKUP,
-                { getBackupObservable() },
-                { view, next -> view.onBackupCompleted() },
-                { view, error -> view.onBackupError(error) })
-
-        startableFirst(RESTORE_BACKUP,
-                { getRestoreObservable() },
-                { view, next -> view.onRestoreCompleted() },
-                { view, error -> view.onRestoreError(error) })
     }
 
     /**
@@ -68,9 +48,13 @@ class BackupPresenter : BasePresenter<BackupFragment>() {
      * @param file the path where the file will be saved.
      */
     fun createBackup(file: File) {
-        if (isUnsubscribed(CREATE_BACKUP)) {
+        if (isUnsubscribed(backupSubscription)) {
-            backupFile = file
+            backupSubscription = getBackupObservable(file)
-            start(CREATE_BACKUP)
+                    .subscribeOn(Schedulers.io())
+                    .observeOn(AndroidSchedulers.mainThread())
+                    .subscribeFirst(
+                            { view, result -> view.onBackupCompleted(file) },
+                            { view, error -> view.onBackupError(error) })
         }
     }
 
@@ -80,30 +64,30 @@ class BackupPresenter : BasePresenter<BackupFragment>() {
      * @param stream the input stream of the backup file.
      */
     fun restoreBackup(stream: InputStream) {
-        if (isUnsubscribed(RESTORE_BACKUP)) {
+        if (isUnsubscribed(restoreSubscription)) {
-            restoreStream = stream
+            restoreSubscription = getRestoreObservable(stream)
-            start(RESTORE_BACKUP)
+                    .subscribeOn(Schedulers.io())
+                    .observeOn(AndroidSchedulers.mainThread())
+                    .subscribeFirst(
+                            { view, result -> view.onRestoreCompleted() },
+                            { view, error -> view.onRestoreError(error) })
         }
     }
 
     /**
      * Returns the observable to save a backup.
      */
-    private fun getBackupObservable(): Observable<Boolean> {
+    private fun getBackupObservable(file: File) = Observable.fromCallable {
-        return Observable.fromCallable {
+        backupManager.backupToFile(file)
-            backupManager.backupToFile(backupFile!!)
+        true
-            true
-        }.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
     }
 
     /**
      * Returns the observable to restore a backup.
      */
-    private fun getRestoreObservable(): Observable<Boolean> {
+    private fun getRestoreObservable(stream: InputStream) = Observable.fromCallable {
-        return Observable.fromCallable {
+        backupManager.restoreFromStream(stream)
-            backupManager.restoreFromStream(restoreStream!!)
+        true
-            true
-        }.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
     }
 
 }

app/src/main/java/eu/kanade/tachiyomi/ui/base/presenter/BasePresenter.kt

@@ -8,13 +8,34 @@ open class BasePresenter<V : ViewWithPresenter<*>> : RxPresenter<V>() {
 
     lateinit var context: Context
 
+    /**
+     * Subscribes an observable with [deliverFirst] and adds it to the presenter's lifecycle
+     * subscription list.
+     *
+     * @param onNext function to execute when the observable emits an item.
+     * @param onError function to execute when the observable throws an error.
+     */
     fun <T> Observable<T>.subscribeFirst(onNext: (V, T) -> Unit, onError: ((V, Throwable) -> Unit)? = null)
-            = compose(deliverFirst<T>()).subscribe(split(onNext, onError))
+            = compose(deliverFirst<T>()).subscribe(split(onNext, onError)).apply { add(this) }
 
+    /**
+     * Subscribes an observable with [deliverLatestCache] and adds it to the presenter's lifecycle
+     * subscription list.
+     *
+     * @param onNext function to execute when the observable emits an item.
+     * @param onError function to execute when the observable throws an error.
+     */
     fun <T> Observable<T>.subscribeLatestCache(onNext: (V, T) -> Unit, onError: ((V, Throwable) -> Unit)? = null)
-            = compose(deliverLatestCache<T>()).subscribe(split(onNext, onError))
+            = compose(deliverLatestCache<T>()).subscribe(split(onNext, onError)).apply { add(this) }
 
+    /**
+     * Subscribes an observable with [deliverReplay] and adds it to the presenter's lifecycle
+     * subscription list.
+     *
+     * @param onNext function to execute when the observable emits an item.
+     * @param onError function to execute when the observable throws an error.
+     */
     fun <T> Observable<T>.subscribeReplay(onNext: (V, T) -> Unit, onError: ((V, Throwable) -> Unit)? = null)
-            = compose(deliverReplay<T>()).subscribe(split(onNext, onError))
+            = compose(deliverReplay<T>()).subscribe(split(onNext, onError)).apply { add(this) }
 
 }

app/src/main/java/eu/kanade/tachiyomi/ui/base/presenter/RxPresenter.java

@@ -113,7 +113,16 @@ public class RxPresenter<View> extends Presenter<View> {
      * @return true if the subscription is null or unsubscribed, false otherwise.
      */
     public boolean isUnsubscribed(int restartableId) {
-        Subscription subscription = restartableSubscriptions.get(restartableId);
+        return isUnsubscribed(restartableSubscriptions.get(restartableId));
+    }
+
+    /**
+     * Checks if a subscription is unsubscribed.
+     *
+     * @param subscription the subscription to check.
+     * @return true if the subscription is null or unsubscribed, false otherwise.
+     */
+    public boolean isUnsubscribed(@Nullable Subscription subscription) {
         return subscription == null || subscription.isUnsubscribed();
     }
 

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

@@ -49,8 +49,8 @@ class MangaPresenter : BasePresenter<MangaActivity>() {
         // Prepare a subject to communicate the chapters and info presenters for the chapter count.
         SharedData.put(ChapterCountEvent())
 
-        add(Observable.just(manga)
+        Observable.just(manga)
-                .subscribeLatestCache({ view, manga -> view.onSetManga(manga) }))
+                .subscribeLatestCache({ view, manga -> view.onSetManga(manga) })
     }
 
     override fun onDestroy() {

app/src/main/java/eu/kanade/tachiyomi/ui/manga/chapter/ChaptersPresenter.kt

@@ -69,8 +69,8 @@ class ChaptersPresenter : BasePresenter<ChaptersFragment>() {
                 { view, error -> Timber.e(error.cause, error.message) })
 
         manga = SharedData.get(MangaEvent::class.java)?.manga ?: return
-        add(Observable.just(manga)
+        Observable.just(manga)
-                .subscribeLatestCache({ view, manga -> view.onNextManga(manga) }))
+                .subscribeLatestCache({ view, manga -> view.onNextManga(manga) })
 
         source = sourceManager.get(manga.source)!!
         start(DB_CHAPTERS)

app/src/main/java/eu/kanade/tachiyomi/ui/manga/info/MangaInfoPresenter.kt

@@ -79,9 +79,9 @@ class MangaInfoPresenter : BasePresenter<MangaInfoFragment>() {
 
         // Update chapter count
         SharedData.get(ChapterCountEvent::class.java)?.let {
-            add(it.observable
+            it.observable
                     .observeOn(AndroidSchedulers.mainThread())
-                    .subscribeLatestCache({ view, count -> view.setChapterCount(count) }))
+                    .subscribeLatestCache({ view, count -> view.setChapterCount(count) })
         }
     }