Merge branch 'master' of github.com:Netflix/RxJava into scala-additions

Author: samuelgruetter

gradle.properties

@@ -1 +1 @@
-version=0.17.0-RC4-SNAPSHOT
+version=0.17.0-RC5-SNAPSHOT

rxjava-contrib/rxjava-debug/src/main/java/rx/operators/DebugSubscriber.java

@@ -4,47 +4,72 @@
 import rx.Observer;
 import rx.Subscriber;
 import rx.functions.Action1;
+import rx.functions.Action2;
 import rx.functions.Func1;
 import rx.plugins.DebugNotification;
 
-public final class DebugSubscriber<T> extends Subscriber<T> {
+public final class DebugSubscriber<T, C> extends Subscriber<T> {
     private final Func1<T, T> onNextHook;
-    final Action1<DebugNotification> events;
-    final Observer<? super T> o;
-    Operator<? extends T, ?> from = null;
-    Operator<?, ? super T> to = null;
+    private final Func1<DebugNotification, C> start;
+    private final Action1<C> complete;
+    private final Action2<C, Throwable> error;
+    private final Observer<? super T> o;
+    private Operator<? extends T, ?> from = null;
+    private Operator<?, ? super T> to = null;
 
     public DebugSubscriber(
             Func1<T, T> onNextHook,
-            Action1<DebugNotification> _events,
+            Func1<DebugNotification, C> start,
+            Action1<C> complete,
+            Action2<C, Throwable> error,
             Subscriber<? super T> _o,
             Operator<? extends T, ?> _out,
             Operator<?, ? super T> _in) {
         super(_o);
-        this.events = _events;
+        this.start = start;
+        this.complete = complete;
+        this.error = error;
         this.o = _o;
         this.onNextHook = onNextHook;
         this.from = _out;
         this.to = _in;
-        this.add(new DebugSubscription<T>(this));
+        this.add(new DebugSubscription<T, C>(this, start, complete, error));
     }
 
     @Override
     public void onCompleted() {
-        events.call(DebugNotification.createOnCompleted(o, from, to));
-        o.onCompleted();
+        final DebugNotification<T, C> n = DebugNotification.createOnCompleted(o, from, to);
+        C context = start.call(n);
+        try {
+            o.onCompleted();
+            complete.call(context);
+        } catch (Throwable e) {
+            error.call(context, e);
+        }
     }
 
     @Override
     public void onError(Throwable e) {
-        events.call(DebugNotification.createOnError(o, from, e, to));
-        o.onError(e);
+        final DebugNotification<T, C> n = DebugNotification.createOnError(o, from, e, to);
+        C context = start.call(n);
+        try {
+            o.onError(e);
+            complete.call(context);
+        } catch (Throwable e2) {
+            error.call(context, e2);
+        }
     }
 
     @Override
     public void onNext(T t) {
-        events.call(DebugNotification.createOnNext(o, from, t, to));
-        o.onNext(onNextHook.call(t));
+        final DebugNotification<T, C> n = DebugNotification.createOnNext(o, from, t, to);
+        C context = start.call(n);
+        try {
+            o.onNext(onNextHook.call(t));
+            complete.call(context);
+        } catch (Throwable e) {
+            error.call(context, e);
+        }
     }
 
     public Operator<? extends T, ?> getFrom() {

rxjava-contrib/rxjava-debug/src/main/java/rx/operators/DebugSubscription.java

@@ -1,19 +1,34 @@
 package rx.operators;
 
 import rx.Subscription;
+import rx.functions.Action1;
+import rx.functions.Action2;
+import rx.functions.Func1;
 import rx.plugins.DebugNotification;
 
-final class DebugSubscription<T> implements Subscription {
-    private final DebugSubscriber<T> debugObserver;
+final class DebugSubscription<T, C> implements Subscription {
+    private final DebugSubscriber<T, C> debugObserver;
+    private final Func1<DebugNotification, C> start;
+    private final Action1<C> complete;
+    private final Action2<C, Throwable> error;
 
-    DebugSubscription(DebugSubscriber<T> debugObserver) {
+    DebugSubscription(DebugSubscriber<T, C> debugObserver, Func1<DebugNotification, C> start, Action1<C> complete, Action2<C, Throwable> error) {
         this.debugObserver = debugObserver;
+        this.start = start;
+        this.complete = complete;
+        this.error = error;
     }
 
     @Override
     public void unsubscribe() {
-        debugObserver.events.call(DebugNotification.<T> createUnsubscribe(debugObserver.o, debugObserver.from, debugObserver.to));
-        debugObserver.unsubscribe();
+        final DebugNotification<T, C> n = DebugNotification.<T, C> createUnsubscribe(debugObserver.getActual(), debugObserver.getFrom(), debugObserver.getTo());
+        C context = start.call(n);
+        try {
+            debugObserver.unsubscribe();
+            complete.call(context);
+        } catch (Throwable e) {
+            error.call(context, e);
+        }
     }
 
     @Override

rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/DebugHook.java

@@ -6,6 +6,7 @@
 import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action1;
+import rx.functions.Action2;
 import rx.functions.Actions;
 import rx.functions.Func1;
 import rx.functions.Functions;
@@ -17,9 +18,11 @@
  * 
  * @author gscampbell
  */
-public class DebugHook extends RxJavaObservableExecutionHook {
+public class DebugHook<C> extends RxJavaObservableExecutionHook {
     private final Func1 onNextHook;
-    private final Action1<DebugNotification> events;
+    private final Func1<DebugNotification, C> start;
+    private final Action1<C> complete;
+    private final Action2<C, Throwable> error;
 
     /**
      * Creates a new instance of the DebugHook RxJava plug-in that can be passed into
@@ -31,18 +34,26 @@ public class DebugHook extends RxJavaObservableExecutionHook {
      * @param events
      *            This action is invoked as each notification is generated
      */
-    public DebugHook(Func1 onNextDataHook, Action1<DebugNotification> events) {
+    public DebugHook(Func1 onNextDataHook, Func1<DebugNotification, C> start, Action1<C> complete, Action2<C, Throwable> error) {
+        this.complete = complete;
+        this.error = error;
         this.onNextHook = onNextDataHook == null ? Functions.identity() : onNextDataHook;
-        this.events = events == null ? Actions.empty() : events;
+        this.start = (Func1<DebugNotification, C>) (start == null ? Actions.empty() : start);
     }
 
     @Override
-    public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> f) {
+    public <T> OnSubscribe<T> onSubscribeStart(final Observable<? extends T> observableInstance, final OnSubscribe<T> f) {
         return new OnSubscribe<T>() {
             @Override
             public void call(Subscriber<? super T> o) {
-                events.call(DebugNotification.createSubscribe(o, f));
-                f.call(wrapOutbound(null, o));
+                C context = start.call(DebugNotification.createSubscribe(o, observableInstance, f));
+                try {
+                    f.call(wrapOutbound(null, o));
+                    complete.call(context);
+                }
+                catch(Throwable e) {
+                    error.call(context, e);
+                }
             }
         };
     }
@@ -54,12 +65,7 @@ public <T> Subscription onSubscribeReturn(Observable<? extends T> observableInst
 
     @Override
     public <T> OnSubscribe<T> onCreate(final OnSubscribe<T> f) {
-        return new OnSubscribe<T>() {
-            @Override
-            public void call(Subscriber<? super T> o) {
-                f.call(wrapInbound(null, o));
-            }
-        };
+        return new OnCreateWrapper<T>(f);
     }
 
     @Override
@@ -81,19 +87,36 @@ public <T> Subscription onAdd(Subscriber<T> subscriber, Subscription s) {
     private <R> Subscriber<? super R> wrapOutbound(Operator<? extends R, ?> bind, Subscriber<? super R> o) {
         if (o instanceof DebugSubscriber) {
             if (bind != null)
-                ((DebugSubscriber<R>) o).setFrom(bind);
+                ((DebugSubscriber<R, C>) o).setFrom(bind);
             return o;
         }
-        return new DebugSubscriber<R>(onNextHook, events, o, bind, null);
+        return new DebugSubscriber<R, C>(onNextHook, start, complete, error, o, bind, null);
     }
 
     @SuppressWarnings("unchecked")
     private <T> Subscriber<? super T> wrapInbound(Operator<?, ? super T> bind, Subscriber<? super T> o) {
         if (o instanceof DebugSubscriber) {
             if (bind != null)
-                ((DebugSubscriber<T>) o).setTo(bind);
+                ((DebugSubscriber<T, C>) o).setTo(bind);
             return o;
         }
-        return new DebugSubscriber<T>(onNextHook, events, o, null, bind);
+        return new DebugSubscriber<T, C>(onNextHook, start, complete, error, o, null, bind);
+    }
+
+    public final class OnCreateWrapper<T> implements OnSubscribe<T> {
+        private final OnSubscribe<T> f;
+
+        private OnCreateWrapper(OnSubscribe<T> f) {
+            this.f = f;
+        }
+
+        @Override
+        public void call(Subscriber<? super T> o) {
+            f.call(wrapInbound(null, o));
+        }
+
+        public OnSubscribe<T> getActual() {
+            return f;
+        }
     }
 }