Reactive code for reading activity

app/src/main/java/acr/browser/lightning/activity/BrowserActivity.java

@@ -111,7 +111,7 @@ import acr.browser.lightning.fragment.BookmarksFragment;
 import acr.browser.lightning.fragment.TabsFragment;
 import acr.browser.lightning.object.SearchAdapter;
 import acr.browser.lightning.react.Schedulers;
-import acr.browser.lightning.react.Subscription;
+import acr.browser.lightning.react.Subscriber;
 import acr.browser.lightning.receiver.NetworkReceiver;
 
 import com.anthonycr.grant.PermissionsManager;
@@ -360,7 +360,7 @@ public abstract class BrowserActivity extends ThemableBrowserActivity implements
         }
 
         mTabsManager.initializeTabs(this, getIntent(), isIncognito())
-                .subscribe(new Subscription<Void>() {
+                .subscribe(new Subscriber<Void>() {
                     @Override
                     public void onNext(Void item) {}
 

app/src/main/java/acr/browser/lightning/activity/IncognitoActivity.java

@@ -11,7 +11,7 @@ import android.webkit.CookieSyncManager;
 import acr.browser.lightning.R;
 import acr.browser.lightning.react.Action;
 import acr.browser.lightning.react.Observable;
-import acr.browser.lightning.react.Subscriber;
+import acr.browser.lightning.react.OnSubscribe;
 
 @SuppressWarnings("deprecation")
 public class IncognitoActivity extends BrowserActivity {
@@ -20,13 +20,13 @@ public class IncognitoActivity extends BrowserActivity {
     public Observable<Void> updateCookiePreference() {
         return Observable.create(new Action<Void>() {
             @Override
-            public void onSubscribe(@NonNull Subscriber<Void> subscriber) {
+            public void onSubscribe(@NonNull OnSubscribe<Void> onSubscribe) {
                 CookieManager cookieManager = CookieManager.getInstance();
                 if (Build.VERSION.SDK_INT < Build.VERSION_CODES.LOLLIPOP) {
                     CookieSyncManager.createInstance(IncognitoActivity.this);
                 }
                 cookieManager.setAcceptCookie(mPreferences.getIncognitoCookiesEnabled());
-                subscriber.onComplete();
+                onSubscribe.onComplete();
             }
         });
     }

app/src/main/java/acr/browser/lightning/activity/MainActivity.java

@@ -11,7 +11,7 @@ import android.webkit.CookieSyncManager;
 import acr.browser.lightning.R;
 import acr.browser.lightning.react.Action;
 import acr.browser.lightning.react.Observable;
-import acr.browser.lightning.react.Subscriber;
+import acr.browser.lightning.react.OnSubscribe;
 
 @SuppressWarnings("deprecation")
 public class MainActivity extends BrowserActivity {
@@ -20,13 +20,13 @@ public class MainActivity extends BrowserActivity {
     public Observable<Void> updateCookiePreference() {
         return Observable.create(new Action<Void>() {
             @Override
-            public void onSubscribe(@NonNull Subscriber<Void> subscriber) {
+            public void onSubscribe(@NonNull OnSubscribe<Void> onSubscribe) {
                 CookieManager cookieManager = CookieManager.getInstance();
                 if (Build.VERSION.SDK_INT < Build.VERSION_CODES.LOLLIPOP) {
                     CookieSyncManager.createInstance(MainActivity.this);
                 }
                 cookieManager.setAcceptCookie(mPreferences.getCookiesEnabled());
-                subscriber.onComplete();
+                onSubscribe.onComplete();
             }
         });
     }

app/src/main/java/acr/browser/lightning/activity/ReadingActivity.java

@@ -10,6 +10,8 @@ import android.graphics.PorterDuff;
 import android.graphics.drawable.ColorDrawable;
 import android.os.AsyncTask;
 import android.os.Bundle;
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
 import android.support.v7.app.AlertDialog;
 import android.support.v7.app.AppCompatActivity;
 import android.support.v7.widget.Toolbar;
@@ -29,6 +31,12 @@ import acr.browser.lightning.R;
 import acr.browser.lightning.app.BrowserApp;
 import acr.browser.lightning.constant.Constants;
 import acr.browser.lightning.preference.PreferenceManager;
+import acr.browser.lightning.react.Action;
+import acr.browser.lightning.react.Observable;
+import acr.browser.lightning.react.OnSubscribe;
+import acr.browser.lightning.react.Schedulers;
+import acr.browser.lightning.react.Subscriber;
+import acr.browser.lightning.react.Subscription;
 import acr.browser.lightning.reading.HtmlFetcher;
 import acr.browser.lightning.reading.JResult;
 import acr.browser.lightning.utils.ThemeUtils;
@@ -50,7 +58,7 @@ public class ReadingActivity extends AppCompatActivity {
     private String mUrl = null;
     private int mTextSize;
     private ProgressDialog mProgressDialog;
-    private PageLoader mLoaderReference;
+    private Subscription mPageLoaderSubscription;
 
     private static final float XXLARGE = 30.0f;
     private static final float XLARGE = 26.0f;
@@ -141,70 +149,87 @@ public class ReadingActivity extends AppCompatActivity {
         }
         if (getSupportActionBar() != null)
             getSupportActionBar().setTitle(Utils.getDomainName(mUrl));
-        mLoaderReference = new PageLoader(this);
+        mPageLoaderSubscription = loadPage(mUrl).subscribeOn(Schedulers.worker())
-        mLoaderReference.executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR, mUrl);
+                .observeOn(Schedulers.main())
-        return true;
+                .subscribe(new Subscriber<ReaderInfo>() {
-    }
+                    @Override
+                    public void onStart() {
+                        mProgressDialog = new ProgressDialog(ReadingActivity.this);
+                        mProgressDialog.setProgressStyle(ProgressDialog.STYLE_SPINNER);
+                        mProgressDialog.setCancelable(false);
+                        mProgressDialog.setIndeterminate(true);
+                        mProgressDialog.setMessage(getString(R.string.loading));
+                        mProgressDialog.show();
+                    }
 
-    private class PageLoader extends AsyncTask<String, Void, Void> {
+                    @Override
+                    public void onNext(@Nullable ReaderInfo item) {
+                        if (item == null || item.getTitle().isEmpty() || item.getBody().isEmpty()) {
+                            setText(getString(R.string.untitled), getString(R.string.loading_failed));
+                        } else {
+                            setText(item.getTitle(), item.getBody());
+                        }
+                    }
 
-        private final WeakReference<Activity> mActivityReference;
+                    @Override
-        private String mTitleText;
+                    public void onError(@NonNull Throwable throwable) {
-        private String mBodyText;
+                        setText(getString(R.string.untitled), getString(R.string.loading_failed));
+                        if (mProgressDialog != null && mProgressDialog.isShowing()) {
+                            mProgressDialog.dismiss();
+                            mProgressDialog = null;
+                        }
+                    }
 
-        public PageLoader(Activity activity) {
+                    @Override
-            mActivityReference = new WeakReference<>(activity);
+                    public void onComplete() {
-        }
+                        if (mProgressDialog != null && mProgressDialog.isShowing()) {
+                            mProgressDialog.dismiss();
+                            mProgressDialog = null;
+                        }
+                    }
+                });
+        return true;
+    }
 
-        @Override
+    private static Observable<ReaderInfo> loadPage(@NonNull final String url) {
-        protected void onPreExecute() {
+        return Observable.create(new Action<ReaderInfo>() {
-            super.onPreExecute();
+            @Override
-            Activity activity = mActivityReference.get();
+            public void onSubscribe(@NonNull OnSubscribe<ReaderInfo> onSubscribe) {
-            if (activity != null) {
+                HtmlFetcher fetcher = new HtmlFetcher();
-                mProgressDialog = new ProgressDialog(activity);
+                try {
-                mProgressDialog.setProgressStyle(ProgressDialog.STYLE_SPINNER);
+                    JResult result = fetcher.fetchAndExtract(url, 2500, true);
-                mProgressDialog.setCancelable(false);
+                    onSubscribe.onNext(new ReaderInfo(result.getTitle(), result.getText()));
-                mProgressDialog.setIndeterminate(true);
+                } catch (Exception e) {
-                mProgressDialog.setMessage(activity.getString(R.string.loading));
+                    onSubscribe.onError(new Throwable("Encountered exception"));
-                mProgressDialog.show();
+                    e.printStackTrace();
+                } catch (OutOfMemoryError e) {
+                    System.gc();
+                    onSubscribe.onError(new Throwable("Out of memory"));
+                    e.printStackTrace();
+                }
+                onSubscribe.onComplete();
             }
-        }
+        });
+    }
 
-        @Override
+    private static class ReaderInfo {
-        protected Void doInBackground(String... params) {
+        @NonNull private final String mTitleText;
-
+        @NonNull private final String mBodyText;
-            HtmlFetcher fetcher = new HtmlFetcher();
+
-            try {
+        public ReaderInfo(@NonNull String title, @NonNull String body) {
-                JResult result = fetcher.fetchAndExtract(params[0], 2500, true);
+            mTitleText = title;
-                mTitleText = result.getTitle();
+            mBodyText = body;
-                mBodyText = result.getText();
-            } catch (Exception e) {
-                mTitleText = "";
-                mBodyText = "";
-                e.printStackTrace();
-            } catch (OutOfMemoryError e) {
-                System.gc();
-                mTitleText = "";
-                mBodyText = "";
-                e.printStackTrace();
-            }
-            return null;
         }
 
-        @Override
+        @NonNull
-        protected void onPostExecute(Void result) {
+        public String getTitle() {
-            if (mProgressDialog != null && mProgressDialog.isShowing()) {
+            return mTitleText;
-                mProgressDialog.dismiss();
-                mProgressDialog = null;
-            }
-            if (mTitleText.isEmpty() || mBodyText.isEmpty()) {
-                setText(getString(R.string.untitled), getString(R.string.loading_failed));
-            } else {
-                setText(mTitleText, mBodyText);
-            }
-            super.onPostExecute(result);
         }
 
+        @NonNull
+        public String getBody() {
+            return mBodyText;
+        }
     }
 
     private void setText(String title, String body) {
@@ -235,11 +260,12 @@ public class ReadingActivity extends AppCompatActivity {
 
     @Override
     protected void onDestroy() {
+        mPageLoaderSubscription.unsubscribe();
+
         if (mProgressDialog != null && mProgressDialog.isShowing()) {
             mProgressDialog.dismiss();
             mProgressDialog = null;
         }
-        mLoaderReference.cancel(true);
         super.onDestroy();
     }
 

app/src/main/java/acr/browser/lightning/activity/TabsManager.java

@@ -24,9 +24,9 @@ import acr.browser.lightning.R;
 import acr.browser.lightning.constant.Constants;
 import acr.browser.lightning.preference.PreferenceManager;
 import acr.browser.lightning.react.Action;
+import acr.browser.lightning.react.OnSubscribe;
 import acr.browser.lightning.react.Schedulers;
 import acr.browser.lightning.react.Subscriber;
-import acr.browser.lightning.react.Subscription;
 import acr.browser.lightning.utils.FileUtils;
 import acr.browser.lightning.react.Observable;
 import acr.browser.lightning.view.LightningView;
@@ -66,14 +66,14 @@ public class TabsManager {
                                                         final boolean incognito) {
         return Observable.create(new Action<Void>() {
             @Override
-            public void onSubscribe(@NonNull final Subscriber<Void> subscriber) {
+            public void onSubscribe(@NonNull final OnSubscribe<Void> onSubscribe) {
 
 
                 // If incognito, only create one tab, do not handle intent
                 // in order to protect user privacy
                 if (incognito && mTabList.isEmpty()) {
                     newTab(activity, null, true);
-                    subscriber.onComplete();
+                    onSubscribe.onComplete();
                     return;
                 }
 
@@ -85,7 +85,7 @@ public class TabsManager {
                 mTabList.clear();
                 mCurrentTab = null;
                 if (mPreferenceManager.getRestoreLostTabsEnabled()) {
-                    restoreLostTabs(url, activity, subscriber);
+                    restoreLostTabs(url, activity, onSubscribe);
                 }
 
             }
@@ -94,9 +94,9 @@ public class TabsManager {
     }
 
     private void restoreLostTabs(final String url, final Activity activity,
-                                 final Subscriber subscriber) {
+                                 final OnSubscribe onSubscribe) {
         restoreState().subscribeOn(Schedulers.worker())
-                .observeOn(Schedulers.main()).subscribe(new Subscription<Bundle>() {
+                .observeOn(Schedulers.main()).subscribe(new Subscriber<Bundle>() {
             @Override
             public void onNext(Bundle item) {
                 LightningView tab = newTab(activity, "", false);
@@ -127,7 +127,7 @@ public class TabsManager {
                 if (mTabList.size() == 0) {
                     newTab(activity, null, false);
                 }
-                subscriber.onComplete();
+                onSubscribe.onComplete();
             }
         });
     }
@@ -331,17 +331,17 @@ public class TabsManager {
     private Observable<Bundle> restoreState() {
         return Observable.create(new Action<Bundle>() {
             @Override
-            public void onSubscribe(@NonNull Subscriber<Bundle> subscriber) {
+            public void onSubscribe(@NonNull OnSubscribe<Bundle> onSubscribe) {
                 Bundle savedState = FileUtils.readBundleFromStorage(mApp, BUNDLE_STORAGE);
                 if (savedState != null) {
                     Log.d(Constants.TAG, "Restoring previous WebView state now");
                     for (String key : savedState.keySet()) {
                         if (key.startsWith(BUNDLE_KEY)) {
-                            subscriber.onNext(savedState.getBundle(key));
+                            onSubscribe.onNext(savedState.getBundle(key));
                         }
                     }
                 }
-                subscriber.onComplete();
+                onSubscribe.onComplete();
             }
         });
     }

app/src/main/java/acr/browser/lightning/react/Action.java

@@ -4,13 +4,13 @@ import android.support.annotation.NonNull;
 
 public interface Action<T> {
     /**
-     * Should be overridden to send the subscriber
+     * Should be overridden to send the onSubscribe
-     * events such as {@link Subscriber#onNext(Object)}
+     * events such as {@link OnSubscribe#onNext(Object)}
-     * or {@link Subscriber#onComplete()}.
+     * or {@link OnSubscribe#onComplete()}.
      *
-     * @param subscriber the subscriber that is sent in
+     * @param onSubscribe the onSubscribe that is sent in
      *                   when the user of the Observable
      *                   subscribes.
      */
-    void onSubscribe(@NonNull Subscriber<T> subscriber);
+    void onSubscribe(@NonNull OnSubscribe<T> onSubscribe);
 }

app/src/main/java/acr/browser/lightning/react/Observable.java

@@ -22,8 +22,8 @@ public class Observable<T> {
     private static final String TAG = Observable.class.getSimpleName();
 
     @NonNull private final Action<T> mAction;
-    @Nullable private Executor mSubscriber;
+    @Nullable private Executor mSubscriberThread;
-    @Nullable private Executor mObserver;
+    @Nullable private Executor mObserverThread;
     @NonNull private final Executor mDefault;
 
     private Observable(@NonNull Action<T> action) {
@@ -49,14 +49,14 @@ public class Observable<T> {
     }
 
     /**
-     * Tells the Observable what Executor that the subscription
+     * Tells the Observable what Executor that the subscriber
      * work should run on.
      *
      * @param subscribeExecutor the Executor to run the work on.
      * @return returns this so that calls can be conveniently chained.
      */
     public Observable<T> subscribeOn(@NonNull Executor subscribeExecutor) {
-        mSubscriber = subscribeExecutor;
+        mSubscriberThread = subscribeExecutor;
         return this;
     }
 
@@ -68,7 +68,7 @@ public class Observable<T> {
      * @return returns this so that calls can be conveniently chained.
      */
     public Observable<T> observeOn(@NonNull Executor observerExecutor) {
-        mObserver = observerExecutor;
+        mObserverThread = observerExecutor;
         return this;
     }
 
@@ -80,10 +80,16 @@ public class Observable<T> {
         executeOnSubscriberThread(new Runnable() {
             @Override
             public void run() {
-                mAction.onSubscribe(new Subscriber<T>() {
+                mAction.onSubscribe(new OnSubscribe<T>(null) {
+                    @Override
+                    public void unsubscribe() {}
+
                     @Override
                     public void onComplete() {}
 
+                    @Override
+                    public void start() {}
+
                     @Override
                     public void onError(@NonNull Throwable throwable) {}
 
@@ -96,123 +102,140 @@ public class Observable<T> {
 
     /**
      * Immediately subscribes to the Observable and starts
-     * sending events from the Observable to the {@link Subscription}.
+     * sending events from the Observable to the {@link Subscriber}.
      *
-     * @param subscription the class that wishes to receive onNext and
+     * @param subscriber the class that wishes to receive onNext and
-     *                     onComplete callbacks from the Observable.
+     *                   onComplete callbacks from the Observable.
      */
-    public void subscribe(@NonNull final Subscription<T> subscription) {
+    public Subscription subscribe(@NonNull Subscriber<T> subscriber) {
-        Preconditions.checkNonNull(subscription);
+
-        executeOnSubscriberThread(new Runnable() {
+        Preconditions.checkNonNull(subscriber);
+
+        final OnSubscribe<T> onSubscribe = new OnSubscribe<T>(subscriber) {
+
+            @Override
+            public void unsubscribe() {
+                setSubscriber(null);
+            }
 
             private boolean mOnCompleteExecuted = false;
 
             @Override
-            public void run() {
+            public void onComplete() {
+                Subscriber<T> subscription = getSubscriber();
+                if (!mOnCompleteExecuted && subscription != null) {
+                    mOnCompleteExecuted = true;
+                    executeOnObserverThread(new OnCompleteRunnable<>(subscription));
+                } else {
+                    Log.e(TAG, "onComplete called more than once");
+                    throw new RuntimeException("onComplete called more than once");
+                }
+            }
+
+            @Override
+            public void start() {
+                Subscriber<T> subscription = getSubscriber();
                 executeOnObserverThread(new OnStartRunnable<>(subscription));
-                mAction.onSubscribe(new Subscriber<T>() {
+            }
-                    @Override
-                    public void onComplete() {
-                        if (!mOnCompleteExecuted) {
-                            mOnCompleteExecuted = true;
-                            executeOnObserverThread(new OnCompleteRunnable<>(subscription));
-                        } else {
-                            Log.e(TAG, "onComplete called more than once");
-                            throw new RuntimeException("onComplete called more than once");
-                        }
-                    }
 
-                    @Override
+            @Override
-                    public void onError(@NonNull final Throwable throwable) {
+            public void onError(@NonNull final Throwable throwable) {
-                        if (!mOnCompleteExecuted) {
+                Subscriber<T> subscription = getSubscriber();
-                            mOnCompleteExecuted = true;
+                if (!mOnCompleteExecuted && subscription != null) {
-                            executeOnObserverThread(new OnErrorRunnable<>(subscription, throwable));
+                    mOnCompleteExecuted = true;
-                        } else {
+                    executeOnObserverThread(new OnErrorRunnable<>(subscription, throwable));
-                            Log.e(TAG, "onComplete already called");
+                } else {
-                            throw new RuntimeException("onComplete already called");
+                    Log.e(TAG, "onComplete already called");
-                        }
+                    throw new RuntimeException("onComplete already called");
-                    }
+                }
+            }
 
-                    @Override
+            @Override
-                    public void onNext(final T item) {
+            public void onNext(final T item) {
-                        if (!mOnCompleteExecuted) {
+                Subscriber<T> subscription = getSubscriber();
-                            executeOnObserverThread(new OnNextRunnable<>(subscription, item));
+                if (!mOnCompleteExecuted && subscription != null) {
-                        } else {
+                    executeOnObserverThread(new OnNextRunnable<>(subscription, item));
-                            Log.e(TAG, "onComplete has been already called, onNext should not be called");
+                } else {
-                            throw new RuntimeException("onNext should not be called after onComplete has been called");
+                    Log.e(TAG, "onComplete has been already called, onNext should not be called");
-                        }
+                    throw new RuntimeException("onNext should not be called after onComplete has been called");
-                    }
+                }
-                });
+            }
+        };
+        executeOnSubscriberThread(new Runnable() {
 
+            @Override
+            public void run() {
+                mAction.onSubscribe(onSubscribe);
             }
         });
+        return onSubscribe;
     }
 
     private void executeOnObserverThread(@NonNull Runnable runnable) {
-        if (mObserver != null) {
+        if (mObserverThread != null) {
-            mObserver.execute(runnable);
+            mObserverThread.execute(runnable);
         } else {
             mDefault.execute(runnable);
         }
     }
 
     private void executeOnSubscriberThread(@NonNull Runnable runnable) {
-        if (mSubscriber != null) {
+        if (mSubscriberThread != null) {
-            mSubscriber.execute(runnable);
+            mSubscriberThread.execute(runnable);
         } else {
             mDefault.execute(runnable);
         }
     }
 
     private static class OnCompleteRunnable<T> implements Runnable {
-        private final Subscription<T> subscription;
+        private final Subscriber<T> subscriber;
 
-        public OnCompleteRunnable(Subscription<T> subscription) {this.subscription = subscription;}
+        public OnCompleteRunnable(Subscriber<T> subscriber) {this.subscriber = subscriber;}
 
         @Override
         public void run() {
-            subscription.onComplete();
+            subscriber.onComplete();
         }
     }
 
     private static class OnNextRunnable<T> implements Runnable {
-        private final Subscription<T> subscription;
+        private final Subscriber<T> subscriber;
         private final T item;
 
-        public OnNextRunnable(Subscription<T> subscription, T item) {
+        public OnNextRunnable(Subscriber<T> subscriber, T item) {
-            this.subscription = subscription;
+            this.subscriber = subscriber;
             this.item = item;
         }
 
         @Override
         public void run() {
-            subscription.onNext(item);
+            subscriber.onNext(item);
         }
     }
 
     private static class OnErrorRunnable<T> implements Runnable {
-        private final Subscription<T> subscription;
+        private final Subscriber<T> subscriber;
         private final Throwable throwable;
 
-        public OnErrorRunnable(Subscription<T> subscription, Throwable throwable) {
+        public OnErrorRunnable(Subscriber<T> subscriber, Throwable throwable) {
-            this.subscription = subscription;
+            this.subscriber = subscriber;
             this.throwable = throwable;
         }
 
         @Override
         public void run() {
-            subscription.onError(throwable);
+            subscriber.onError(throwable);
         }
     }
 
     private static class OnStartRunnable<T> implements Runnable {
-        private final Subscription<T> subscription;
+        private final Subscriber<T> subscriber;
 
-        public OnStartRunnable(Subscription<T> subscription) {this.subscription = subscription;}
+        public OnStartRunnable(Subscriber<T> subscriber) {this.subscriber = subscriber;}
 
         @Override
         public void run() {
-            subscription.onStart();
+            subscriber.onStart();
         }
     }
 }

app/src/main/java/acr/browser/lightning/react/OnSubscribe.java

@@ -0,0 +1,59 @@
+package acr.browser.lightning.react;
+
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+
+public abstract class OnSubscribe<T> implements Subscription {
+
+    @Nullable private Subscriber<T> mSubscriber;
+
+    public OnSubscribe(@Nullable Subscriber<T> subscriber) {
+        mSubscriber = subscriber;
+        start();
+    }
+
+    public abstract void start();
+
+    @Nullable
+    public Subscriber<T> getSubscriber() {
+        return mSubscriber;
+    }
+
+    public void setSubscriber(@Nullable Subscriber<T> subscriber) {
+        mSubscriber = subscriber;
+    }
+
+    /**
+     * Called when the observable
+     * runs into an error that will
+     * cause it to abort and not finish.
+     * Receiving this callback means that
+     * the observable is dead and no
+     * {@link #onComplete()} or {@link #onNext(Object)}
+     * callbacks will be called.
+     *
+     * @param throwable an optional throwable that could
+     *                  be sent.
+     */
+    public abstract void onError(@NonNull Throwable throwable);
+
+    /**
+     * Called when the Observer emits an
+     * item. It can be called multiple times.
+     * It cannot be called after onComplete
+     * has been called.
+     *
+     * @param item the item that has been emitted,
+     *             can be null.
+     */
+    public abstract void onNext(@Nullable T item);
+
+    /**
+     * This method is called when the observer is
+     * finished sending the subscriber events. It
+     * is guaranteed that no other methods will be
+     * called on the Subscriber after this method
+     * has been called.
+     */
+    public abstract void onComplete();
+}

app/src/main/java/acr/browser/lightning/react/Subscriber.java

@@ -3,7 +3,7 @@ package acr.browser.lightning.react;
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 
-public interface Subscriber<T> {
+public abstract class Subscriber<T> {
 
     /**
      * Called when the observable
@@ -17,7 +17,13 @@ public interface Subscriber<T> {
      * @param throwable an optional throwable that could
      *                  be sent.
      */
-    void onError(@NonNull Throwable throwable);
+    public void onError(@NonNull Throwable throwable) {}
+
+    /**
+     * Called before the observer begins
+     * to process and emit items or complete.
+     */
+    public void onStart() {}
 
     /**
      * Called when the Observer emits an
@@ -28,14 +34,14 @@ public interface Subscriber<T> {
      * @param item the item that has been emitted,
      *             can be null.
      */
-    void onNext(@Nullable T item);
+    public void onNext(@Nullable T item) {}
 
     /**
      * This method is called when the observer is
      * finished sending the subscriber events. It
      * is guaranteed that no other methods will be
-     * called on the Subscription after this method
+     * called on the Subscriber after this method
      * has been called.
      */
-    void onComplete();
+    public void onComplete() {}
 }

app/src/main/java/acr/browser/lightning/react/Subscription.java

@@ -1,47 +1,7 @@
 package acr.browser.lightning.react;
 
-import android.support.annotation.NonNull;
+public interface Subscription {
-import android.support.annotation.Nullable;
 
-public abstract class Subscription<T> {
+    void unsubscribe();
 
-    /**
-     * Called when the observable
-     * runs into an error that will
-     * cause it to abort and not finish.
-     * Receiving this callback means that
-     * the observable is dead and no
-     * {@link #onComplete()} or {@link #onNext(Object)}
-     * callbacks will be called.
-     *
-     * @param throwable an optional throwable that could
-     *                  be sent.
-     */
-    public void onError(@NonNull Throwable throwable) {}
-
-    /**
-     * Called before the observer begins
-     * to process and emit items or complete.
-     */
-    public void onStart() {}
-
-    /**
-     * Called when the Observer emits an
-     * item. It can be called multiple times.
-     * It cannot be called after onComplete
-     * has been called.
-     *
-     * @param item the item that has been emitted,
-     *             can be null.
-     */
-    public void onNext(@Nullable T item) {}
-
-    /**
-     * This method is called when the observer is
-     * finished sending the subscriber events. It
-     * is guaranteed that no other methods will be
-     * called on the Subscription after this method
-     * has been called.
-     */
-    public void onComplete() {}
 }

app/src/main/java/acr/browser/lightning/view/LightningView.java

@@ -55,8 +55,8 @@ import acr.browser.lightning.preference.PreferenceManager;
 import acr.browser.lightning.react.Action;
 import acr.browser.lightning.react.Observable;
 import acr.browser.lightning.react.Schedulers;
+import acr.browser.lightning.react.OnSubscribe;
 import acr.browser.lightning.react.Subscriber;
-import acr.browser.lightning.react.Subscription;
 import acr.browser.lightning.utils.ProxyUtils;
 import acr.browser.lightning.utils.ThemeUtils;
 import acr.browser.lightning.utils.UrlUtils;
@@ -375,7 +375,7 @@ public class LightningView {
 
         getPathObservable("appcache")
                 .subscribeOn(Schedulers.worker())
-                .subscribe(new Subscription<File>() {
+                .subscribe(new Subscriber<File>() {
                     @Override
                     public void onNext(File item) {
                         settings.setAppCachePath(item.getPath());
@@ -387,7 +387,7 @@ public class LightningView {
 
         getPathObservable("geolocation")
                 .subscribeOn(Schedulers.worker())
-                .subscribe(new Subscription<File>() {
+                .subscribe(new Subscriber<File>() {
                     @Override
                     public void onNext(File item) {
                         settings.setGeolocationDatabasePath(item.getPath());
@@ -399,7 +399,7 @@ public class LightningView {
 
         getPathObservable("databases")
                 .subscribeOn(Schedulers.worker())
-                .subscribe(new Subscription<File>() {
+                .subscribe(new Subscriber<File>() {
                     @Override
                     public void onNext(File item) {
                         if (API < Build.VERSION_CODES.KITKAT) {
@@ -417,10 +417,10 @@ public class LightningView {
     private Observable<File> getPathObservable(final String subFolder) {
         return Observable.create(new Action<File>() {
             @Override
-            public void onSubscribe(@NonNull Subscriber<File> subscriber) {
+            public void onSubscribe(@NonNull OnSubscribe<File> onSubscribe) {
                 File file = BrowserApp.get(mActivity).getDir(subFolder, 0);
-                subscriber.onNext(file);
+                onSubscribe.onNext(file);
-                subscriber.onComplete();
+                onSubscribe.onComplete();
             }
         });
     }