Skip to content

Commit

Permalink
Merge pull request #945 from benjchristensen/merge-940
Browse files Browse the repository at this point in the history 
Manual Merge 940
  • Loading branch information
@benjchristensen
benjchristensen committed Mar 6, 2014
2 parents ec7eb97 + 8c1e037 commit 410960d
Show file tree
Hide file tree
Showing 11 changed files with 641 additions and 227 deletions.
49 changes: 20 additions & 29 deletions rxjava-contrib/rxjava-debug/src/main/java/rx/operators/DebugSubscriber.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,72 +3,63 @@
import rx.Observable.Operator;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Action2;
import rx.functions.Func1;
import rx.plugins.DebugNotification;
import rx.plugins.DebugNotificationListener;

public final class DebugSubscriber<T, C> extends Subscriber<T> {
private final Func1<T, T> onNextHook;
private final Func1<DebugNotification, C> start;
private final Action1<C> complete;
private final Action2<C, Throwable> error;
private DebugNotificationListener<C> listener;
private final Observer<? super T> o;
private Operator<? extends T, ?> from = null;
private Operator<?, ? super T> to = null;

public DebugSubscriber(
Func1<T, T> onNextHook,
Func1<DebugNotification, C> start,
Action1<C> complete,
Action2<C, Throwable> error,
DebugNotificationListener<C> listener,
Subscriber<? super T> _o,
Operator<? extends T, ?> _out,
Operator<?, ? super T> _in) {
super(_o);
this.start = start;
this.complete = complete;
this.error = error;
this.listener = listener;
this.o = _o;
this.onNextHook = onNextHook;
this.from = _out;
this.to = _in;
this.add(new DebugSubscription<T, C>(this, start, complete, error));
this.add(new DebugSubscription<T, C>(this, listener));
}

@Override
public void onCompleted() {
final DebugNotification<T, C> n = DebugNotification.createOnCompleted(o, from, to);
C context = start.call(n);
final DebugNotification<T> n = DebugNotification.createOnCompleted(o, from, to);
C context = listener.start(n);
try {
o.onCompleted();
complete.call(context);
listener.complete(context);
} catch (Throwable e) {
error.call(context, e);
listener.error(context, e);
}
}

@Override
public void onError(Throwable e) {
final DebugNotification<T, C> n = DebugNotification.createOnError(o, from, e, to);
C context = start.call(n);
final DebugNotification<T> n = DebugNotification.createOnError(o, from, e, to);
C context = listener.start(n);
try {
o.onError(e);
complete.call(context);
listener.complete(context);
} catch (Throwable e2) {
error.call(context, e2);
listener.error(context, e2);
}
}

@Override
public void onNext(T t) {
final DebugNotification<T, C> n = DebugNotification.createOnNext(o, from, t, to);
C context = start.call(n);
final DebugNotification<T> n = DebugNotification.createOnNext(o, from, t, to);
t = (T) listener.onNext(n);

C context = listener.start(n);
try {
o.onNext(onNextHook.call(t));
complete.call(context);
o.onNext(t);
listener.complete(context);
} catch (Throwable e) {
error.call(context, e);
listener.error(context, e);
}
}

Expand Down
22 changes: 8 additions & 14 deletions rxjava-contrib/rxjava-debug/src/main/java/rx/operators/DebugSubscription.java
View file
Original file line number Diff line number Diff line change
@@ -1,33 +1,27 @@
package rx.operators;

import rx.Subscription;
import rx.functions.Action1;
import rx.functions.Action2;
import rx.functions.Func1;
import rx.plugins.DebugNotification;
import rx.plugins.DebugNotificationListener;

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;
private DebugNotificationListener<C> listener;

DebugSubscription(DebugSubscriber<T, C> debugObserver, Func1<DebugNotification, C> start, Action1<C> complete, Action2<C, Throwable> error) {
DebugSubscription(DebugSubscriber<T, C> debugObserver, DebugNotificationListener<C> listener) {
this.debugObserver = debugObserver;
this.start = start;
this.complete = complete;
this.error = error;
this.listener = listener;
}

@Override
public void unsubscribe() {
final DebugNotification<T, C> n = DebugNotification.<T, C> createUnsubscribe(debugObserver.getActual(), debugObserver.getFrom(), debugObserver.getTo());
C context = start.call(n);
final DebugNotification<T> n = DebugNotification.<T> createUnsubscribe(debugObserver.getActual(), debugObserver.getFrom(), debugObserver.getTo());
C context = listener.start(n);
try {
debugObserver.unsubscribe();
complete.call(context);
listener.complete(context);
} catch (Throwable e) {
error.call(context, e);
listener.error(context, e);
}
}

Expand Down
73 changes: 34 additions & 39 deletions rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/DebugHook.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,7 @@
* @author gscampbell
*/
public class DebugHook<C> extends RxJavaObservableExecutionHook {
private final Func1 onNextHook;
private final Func1<DebugNotification, C> start;
private final Action1<C> complete;
private final Action2<C, Throwable> error;
private DebugNotificationListener<C> listener;

/**
* Creates a new instance of the DebugHook RxJava plug-in that can be passed into
Expand All @@ -34,39 +31,59 @@ public class DebugHook<C> extends RxJavaObservableExecutionHook {
* @param events
* This action is invoked as each notification is generated
*/
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.start = (Func1<DebugNotification, C>) (start == null ? Actions.empty() : start);
public DebugHook(DebugNotificationListener<C> listener) {
if (listener == null)
throw new IllegalArgumentException("The debug listener must not be null");
this.listener = listener;
}

@Override
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) {
C context = start.call(DebugNotification.createSubscribe(o, observableInstance, f));
final DebugNotification<T> n = DebugNotification.createSubscribe(o, observableInstance, f);
o = wrapOutbound(null, o);

C context = listener.start(n);
try {
f.call(wrapOutbound(null, o));
complete.call(context);
f.call(o);
listener.complete(context);
}
catch(Throwable e) {
error.call(context, e);
listener.error(context, e);
}
}
};
}

@Override
public <T> Subscription onSubscribeReturn(Observable<? extends T> observableInstance, Subscription subscription) {
public <T> Subscription onSubscribeReturn(Subscription subscription) {
return subscription;
}

@Override
public <T> OnSubscribe<T> onCreate(final OnSubscribe<T> f) {
return new OnCreateWrapper<T>(f);
return new DebugOnSubscribe<T>(f);
}

public final class DebugOnSubscribe<T> implements OnSubscribe<T> {
private final OnSubscribe<T> f;

private DebugOnSubscribe(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;
}
}


@Override
public <T, R> Operator<? extends R, ? super T> onLift(final Operator<? extends R, ? super T> bind) {
Expand All @@ -78,19 +95,14 @@ public Subscriber<? super T> call(final Subscriber<? super R> o) {
};
}

@Override
public <T> Subscription onAdd(Subscriber<T> subscriber, Subscription s) {
return s;
}

@SuppressWarnings("unchecked")
private <R> Subscriber<? super R> wrapOutbound(Operator<? extends R, ?> bind, Subscriber<? super R> o) {
if (o instanceof DebugSubscriber) {
if (bind != null)
((DebugSubscriber<R, C>) o).setFrom(bind);
return o;
}
return new DebugSubscriber<R, C>(onNextHook, start, complete, error, o, bind, null);
return new DebugSubscriber<R, C>(listener, o, bind, null);
}

@SuppressWarnings("unchecked")
Expand All @@ -100,23 +112,6 @@ private <T> Subscriber<? super T> wrapInbound(Operator<?, ? super T> bind, Subsc
((DebugSubscriber<T, C>) o).setTo(bind);
return o;
}
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;
}
return new DebugSubscriber<T, C>(listener, o, null, bind);
}
}
Loading

0 comments on commit 410960d

Please sign in to comment.