Add support for onStart and onError

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

@@ -84,6 +84,9 @@ public class Observable<T> {
                     @Override
                     public void onComplete() {}
 
+                    @Override
+                    public void onError(@NonNull Throwable throwable) {}
+
                     @Override
                     public void onNext(T item) {}
                 });
@@ -106,6 +109,7 @@ public class Observable<T> {
 
             @Override
             public void run() {
+                executeOnObserverThread(new OnStartRunnable<>(subscription));
                 mAction.onSubscribe(new Subscriber<T>() {
                     @Override
                     public void onComplete() {
@@ -118,6 +122,17 @@ public class Observable<T> {
                         }
                     }
 
+                    @Override
+                    public void onError(@NonNull final Throwable throwable) {
+                        if (!mOnCompleteExecuted) {
+                            mOnCompleteExecuted = true;
+                            executeOnObserverThread(new OnErrorRunnable<>(subscription, throwable));
+                        } else {
+                            Log.e(TAG, "onComplete already called");
+                            throw new RuntimeException("onComplete already called");
+                        }
+                    }
+
                     @Override
                     public void onNext(final T item) {
                         if (!mOnCompleteExecuted) {
@@ -174,5 +189,31 @@ public class Observable<T> {
             subscription.onNext(item);
         }
     }
+
+    private static class OnErrorRunnable<T> implements Runnable {
+        private final Subscription<T> subscription;
+        private final Throwable throwable;
+
+        public OnErrorRunnable(Subscription<T> subscription, Throwable throwable) {
+            this.subscription = subscription;
+            this.throwable = throwable;
+        }
+
+        @Override
+        public void run() {
+            subscription.onError(throwable);
+        }
+    }
+
+    private static class OnStartRunnable<T> implements Runnable {
+        private final Subscription<T> subscription;
+
+        public OnStartRunnable(Subscription<T> subscription) {this.subscription = subscription;}
+
+        @Override
+        public void run() {
+            subscription.onStart();
+        }
+    }
 }
 

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

@@ -1,8 +1,24 @@
 package acr.browser.lightning.react;
 
+import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 
 public interface Subscriber<T> {
+
+    /**
+     * 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.
+     */
+    void onError(@NonNull Throwable throwable);
+
     /**
      * Called when the Observer emits an
      * item. It can be called multiple times.

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

@@ -1,8 +1,29 @@
 package acr.browser.lightning.react;
 
+import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 
-public interface Subscription<T> {
+public abstract class Subscription<T> {
+
+    /**
+     * 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
@@ -13,7 +34,7 @@ public interface Subscription<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
@@ -22,5 +43,5 @@ public interface Subscription<T> {
      * called on the Subscription after this method
      * has been called.
      */
-    void onComplete();
+    public void onComplete() {}
 }