Merge pull request #2842 from AntennaPod/rxjava2

Migrate rxjava to version 2

6 files changed, 51 insertions, 34 deletions

diff --git a/core/build.gradle b/core/build.gradle
index 80adefa60..95b9db35a 100644
--- a/core/build.gradle
+++ b/core/build.gradle
@@ -65,7 +65,8 @@ dependencies {
     implementation "com.squareup.okhttp3:okhttp-urlconnection:$okhttpVersion"
     implementation "com.squareup.okio:okio:$okioVersion"
     implementation "de.greenrobot:eventbus:$eventbusVersion"
-    implementation "io.reactivex:rxandroid:$rxAndroidVersion"
+    implementation "io.reactivex.rxjava2:rxandroid:$rxAndroidVersion"
+    implementation "io.reactivex.rxjava2:rxjava:$rxJavaVersion"
     implementation "org.awaitility:awaitility:$awaitilityVersion"
 
     implementation "com.google.android.exoplayer:exoplayer:$exoPlayerVersion"
diff --git a/core/src/main/java/de/danoeh/antennapod/core/storage/DBReader.java b/core/src/main/java/de/danoeh/antennapod/core/storage/DBReader.java
index 718459db8..83aa4c780 100644
--- a/core/src/main/java/de/danoeh/antennapod/core/storage/DBReader.java
+++ b/core/src/main/java/de/danoeh/antennapod/core/storage/DBReader.java
@@ -1,6 +1,7 @@
 package de.danoeh.antennapod.core.storage;
 
 import android.database.Cursor;
+import android.support.annotation.Nullable;
 import android.support.v4.util.ArrayMap;
 import android.text.TextUtils;
 import android.util.Log;
@@ -574,6 +575,7 @@ public final class DBReader {
         }
     }
 
+    @Nullable
     static Feed getFeed(final long feedId, PodDBAdapter adapter) {
         Feed feed = null;
         Cursor cursor = null;
diff --git a/core/src/main/java/de/danoeh/antennapod/core/util/Consumer.java b/core/src/main/java/de/danoeh/antennapod/core/util/Consumer.java
new file mode 100644
index 000000000..13a87af0d
--- /dev/null
+++ b/core/src/main/java/de/danoeh/antennapod/core/util/Consumer.java
@@ -0,0 +1,5 @@
+package de.danoeh.antennapod.core.util;
+
+public interface Consumer<T> {
+    void accept(T t);
+}
diff --git a/core/src/main/java/de/danoeh/antennapod/core/util/Function.java b/core/src/main/java/de/danoeh/antennapod/core/util/Function.java
new file mode 100644
index 000000000..c4f4ad68a
--- /dev/null
+++ b/core/src/main/java/de/danoeh/antennapod/core/util/Function.java
@@ -0,0 +1,7 @@
+package de.danoeh.antennapod.core.util;
+
+import io.reactivex.annotations.NonNull;
+
+public interface Function<T, R> {
+    R apply(@NonNull T t);
+}
diff --git a/core/src/main/java/de/danoeh/antennapod/core/util/NetworkUtils.java b/core/src/main/java/de/danoeh/antennapod/core/util/NetworkUtils.java
index 0120790ed..b34ba196d 100644
--- a/core/src/main/java/de/danoeh/antennapod/core/util/NetworkUtils.java
+++ b/core/src/main/java/de/danoeh/antennapod/core/util/NetworkUtils.java
@@ -18,12 +18,13 @@ import de.danoeh.antennapod.core.feed.FeedMedia;
 import de.danoeh.antennapod.core.preferences.UserPreferences;
 import de.danoeh.antennapod.core.service.download.AntennapodHttpClient;
 import de.danoeh.antennapod.core.storage.DBWriter;
+import io.reactivex.Single;
+import io.reactivex.SingleOnSubscribe;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.schedulers.Schedulers;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
-import rx.Observable;
-import rx.android.schedulers.AndroidSchedulers;
-import rx.schedulers.Schedulers;
 
 public class NetworkUtils {
     private NetworkUtils(){}
@@ -111,11 +112,10 @@ public class NetworkUtils {
         return null;
     }
 
-	public static Observable<Long> getFeedMediaSizeObservable(FeedMedia media) {
-        return Observable.create((Observable.OnSubscribe<Long>) subscriber -> {
+	public static Single<Long> getFeedMediaSizeObservable(FeedMedia media) {
+        return Single.create((SingleOnSubscribe<Long>) emitter -> {
             if (!NetworkUtils.isDownloadAllowed()) {
-                subscriber.onNext(0L);
-                subscriber.onCompleted();
+                emitter.onSuccess(0L);
                 return;
             }
             long size = Integer.MIN_VALUE;
@@ -129,8 +129,7 @@ public class NetworkUtils {
 
                 String url = media.getDownload_url();
                 if(TextUtils.isEmpty(url)) {
-                    subscriber.onNext(0L);
-                    subscriber.onCompleted();
+                    emitter.onSuccess(0L);
                     return;
                 }
 
@@ -150,8 +149,7 @@ public class NetworkUtils {
                         }
                     }
                 } catch (IOException e) {
-                    subscriber.onNext(0L);
-                    subscriber.onCompleted();
+                    emitter.onSuccess(0L);
                     Log.e(TAG, Log.getStackTraceString(e));
                     return; // better luck next time
                 }
@@ -163,11 +161,10 @@ public class NetworkUtils {
             } else {
                 media.setSize(size);
             }
-            subscriber.onNext(size);
-            subscriber.onCompleted();
+            emitter.onSuccess(size);
             DBWriter.setFeedMedia(media);
         })
-                .subscribeOn(Schedulers.newThread())
+                .subscribeOn(Schedulers.io())
 				.observeOn(AndroidSchedulers.mainThread());
     }
 
diff --git a/core/src/main/java/de/danoeh/antennapod/core/util/playback/PlaybackController.java b/core/src/main/java/de/danoeh/antennapod/core/util/playback/PlaybackController.java
index 408789cc9..847c5223b 100644
--- a/core/src/main/java/de/danoeh/antennapod/core/util/playback/PlaybackController.java
+++ b/core/src/main/java/de/danoeh/antennapod/core/util/playback/PlaybackController.java
@@ -38,11 +38,12 @@ import de.danoeh.antennapod.core.service.playback.PlayerStatus;
 import de.danoeh.antennapod.core.storage.DBTasks;
 import de.danoeh.antennapod.core.util.Converter;
 import de.danoeh.antennapod.core.util.playback.Playable.PlayableUtils;
-import rx.Observable;
-import rx.Single;
-import rx.Subscription;
-import rx.android.schedulers.AndroidSchedulers;
-import rx.schedulers.Schedulers;
+import io.reactivex.Maybe;
+import io.reactivex.MaybeOnSubscribe;
+import io.reactivex.Observable;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.schedulers.Schedulers;
 
 /**
  * Communicates with the playback service. GUI classes should use this class to
@@ -70,7 +71,8 @@ public abstract class PlaybackController {
     private boolean released = false;
     private boolean initialized = false;
 
-    private Subscription serviceBinder;
+    private Disposable serviceBinder;
+    private Disposable mediaLoader;
 
     /**
      * True if controller should reinit playback service if 'pause' button is
@@ -145,7 +147,7 @@ public abstract class PlaybackController {
         }
 
         if(serviceBinder != null) {
-            serviceBinder.unsubscribe();
+            serviceBinder.dispose();
         }
         try {
             activity.unbindService(mConnection);
@@ -180,10 +182,10 @@ public abstract class PlaybackController {
     private void bindToService() {
         Log.d(TAG, "Trying to connect to service");
         if (serviceBinder != null) {
-            serviceBinder.unsubscribe();
+            serviceBinder.dispose();
         }
         serviceBinder = Observable.fromCallable(this::getPlayLastPlayedMediaIntent)
-                .subscribeOn(Schedulers.newThread())
+                .subscribeOn(Schedulers.io())
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(intent -> {
                     boolean bound = false;
@@ -777,15 +779,18 @@ public abstract class PlaybackController {
     }
 
     private void initServiceNotRunning() {
-        Single.create(subscriber -> subscriber.onSuccess(getMedia()))
-            .subscribeOn(Schedulers.newThread())
+        mediaLoader = Maybe.create((MaybeOnSubscribe<Playable>) emitter -> {
+            Playable media = getMedia();
+            if (media != null) {
+                emitter.onSuccess(media);
+            } else {
+                emitter.onComplete();
+            }
+        })
+            .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
-            .subscribe((Object media) -> {
-                if (media == null) {
-                    return;
-                }
-
-                if (((Playable) media).getMediaType() == MediaType.AUDIO) {
+            .subscribe(media -> {
+                if (media.getMediaType() == MediaType.AUDIO) {
                     TypedArray res = activity.obtainStyledAttributes(new int[]{
                             de.danoeh.antennapod.core.R.attr.av_play_big});
                     getPlayButton().setImageResource(
@@ -794,7 +799,7 @@ public abstract class PlaybackController {
                 } else {
                     getPlayButton().setImageResource(R.drawable.ic_av_play_circle_outline_80dp);
                 }
-            });
+            }, error -> Log.e(TAG, Log.getStackTraceString(error)));
     }
 
     /**
@@ -802,7 +807,7 @@ public abstract class PlaybackController {
      */
     public class MediaPositionObserver implements Runnable {
 
-        public static final int WAITING_INTERVALL = 1000;
+        static final int WAITING_INTERVALL = 1000;
 
         @Override
         public void run() {