Migrate rxjava to version 2

6 files changed, 60 insertions, 48 deletions

diff --git a/core/src/main/java/de/danoeh/antennapod/core/preferences/UserPreferences.java b/core/src/main/java/de/danoeh/antennapod/core/preferences/UserPreferences.java
index 6aaf65e18..58071495a 100644
--- a/core/src/main/java/de/danoeh/antennapod/core/preferences/UserPreferences.java
+++ b/core/src/main/java/de/danoeh/antennapod/core/preferences/UserPreferences.java
@@ -8,14 +8,7 @@ import android.support.annotation.NonNull;
 import android.support.v4.app.NotificationCompat;
 import android.text.TextUtils;
 import android.util.Log;
-import de.danoeh.antennapod.core.R;
-import de.danoeh.antennapod.core.service.download.ProxyConfig;
-import de.danoeh.antennapod.core.storage.APCleanupAlgorithm;
-import de.danoeh.antennapod.core.storage.APNullCleanupAlgorithm;
-import de.danoeh.antennapod.core.storage.APQueueCleanupAlgorithm;
-import de.danoeh.antennapod.core.storage.EpisodeCleanupAlgorithm;
-import de.danoeh.antennapod.core.util.Converter;
-import de.danoeh.antennapod.core.util.download.AutoUpdateManager;
+
 import org.json.JSONArray;
 import org.json.JSONException;
 
@@ -27,6 +20,15 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import de.danoeh.antennapod.core.R;
+import de.danoeh.antennapod.core.service.download.ProxyConfig;
+import de.danoeh.antennapod.core.storage.APCleanupAlgorithm;
+import de.danoeh.antennapod.core.storage.APNullCleanupAlgorithm;
+import de.danoeh.antennapod.core.storage.APQueueCleanupAlgorithm;
+import de.danoeh.antennapod.core.storage.EpisodeCleanupAlgorithm;
+import de.danoeh.antennapod.core.util.Converter;
+import de.danoeh.antennapod.core.util.download.AutoUpdateManager;
+
 /**
  * Provides access to preferences set by the user in the settings screen. A
  * private instance of this class must first be instantiated via
@@ -108,6 +110,7 @@ public class UserPreferences {
     private static final String PREF_RIGHT_VOLUME = "prefRightVolume";
 
     // Experimental
+    public static final String PREF_SONIC = "prefSonic";
     private static final String PREF_STEREO_TO_MONO = "PrefStereoToMono";
     public static final String PREF_CAST_ENABLED = "prefCast"; //Used for enabling Chromecast support
     public static final int EPISODE_CLEANUP_QUEUE = -1;
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 456d05ded..9b574be47 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..f6893315f
--- /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);
+}
\ No newline at end of file
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 cbda583fa..3349ed176 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,13 +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.Subscriber;
-import rx.android.schedulers.AndroidSchedulers;
-import rx.schedulers.Schedulers;
 
 public class NetworkUtils {
 
@@ -111,11 +111,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 +128,7 @@ public class NetworkUtils {
 
                 String url = media.getDownload_url();
                 if(TextUtils.isEmpty(url)) {
-                    subscriber.onNext(0L);
-                    subscriber.onCompleted();
+                    emitter.onSuccess(0L);
                     return;
                 }
 
@@ -150,8 +148,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 +160,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 31067839a..036ca446c 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
@@ -7,12 +7,10 @@ import android.content.Context;
 import android.content.Intent;
 import android.content.IntentFilter;
 import android.content.ServiceConnection;
-import android.content.SharedPreferences;
 import android.content.res.TypedArray;
 import android.media.MediaPlayer;
 import android.os.Build;
 import android.os.IBinder;
-import android.preference.PreferenceManager;
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 import android.support.v4.content.ContextCompat;
@@ -24,10 +22,8 @@ import android.widget.ImageButton;
 import android.widget.SeekBar;
 import android.widget.TextView;
 
-import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 import de.danoeh.antennapod.core.R;
@@ -42,13 +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.Completable;
-import rx.Observable;
-import rx.Single;
-import rx.Subscription;
-import rx.android.schedulers.AndroidSchedulers;
-import rx.observers.Subscribers;
-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
@@ -76,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
@@ -151,7 +147,7 @@ public abstract class PlaybackController {
         }
 
         if(serviceBinder != null) {
-            serviceBinder.unsubscribe();
+            serviceBinder.dispose();
         }
         try {
             activity.unbindService(mConnection);
@@ -186,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;
@@ -783,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(
@@ -800,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)));
     }
 
     /**
@@ -808,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() {