@Test(expected = OnErrorNotImplementedException.class) public void testThrowDuringOnNext() throws Exception { PublishSubject<Object> view = PublishSubject.create(); final PublishSubject<Integer> subject = PublishSubject.create(); new DeliverFirst<Object, Integer>(view) .call(subject) .subscribe(new Action1<Delivery<Object, Integer>>() { @Override public void call(Delivery<Object, Integer> delivery) { delivery.split( new Action2<Object, Integer>() { @Override public void call(Object o, Integer integer) { throw new RuntimeException(); } }, null ); } }); subject.onNext(3); view.onNext(100); }
@Override public Observable<Delivery<View, T>> call(Observable<T> observable) { return observable.materialize() .take(1) .switchMap(new Func1<Notification<T>, Observable<? extends Delivery<View, T>>>() { @Override public Observable<? extends Delivery<View, T>> call(final Notification<T> notification) { return view.map(new Func1<View, Delivery<View, T>>() { @Override public Delivery<View, T> call(View view) { return view == null ? null : new Delivery<>(view, notification); } }); } }) .filter(new Func1<Delivery<View, T>, Boolean>() { @Override public Boolean call(Delivery<View, T> delivery) { return delivery != null; } }) .take(1); }
DeliverFirst implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { return observable.materialize() .take(1) .switchMap(new Func1<Notification<T>, Observable<? extends Delivery<View, T>>>() { @Override public Observable<? extends Delivery<View, T>> call(final Notification<T> notification) { return view.map(new Func1<View, Delivery<View, T>>() { @Override public Delivery<View, T> call(View view) { return view == null ? null : new Delivery<>(view, notification); } }); } }) .filter(new Func1<Delivery<View, T>, Boolean>() { @Override public Boolean call(Delivery<View, T> delivery) { return delivery != null; } }) .take(1); } }
DeliverFirst implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { return observable.materialize() .take(1) .switchMap(new Func1<Notification<T>, Observable<? extends Delivery<View, T>>>() { @Override public Observable<? extends Delivery<View, T>> call(final Notification<T> notification) { return view.map(new Func1<View, Delivery<View, T>>() { @Override public Delivery<View, T> call(View view) { return view == null ? null : new Delivery<>(view, notification); } }); } }) .filter(new Func1<Delivery<View, T>, Boolean>() { @Override public Boolean call(Delivery<View, T> delivery) { return delivery != null; } }) .take(1); } DeliverFirst(Observable<View> view); }
DeliverFirst implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { return observable.materialize() .take(1) .switchMap(new Func1<Notification<T>, Observable<? extends Delivery<View, T>>>() { @Override public Observable<? extends Delivery<View, T>> call(final Notification<T> notification) { return view.map(new Func1<View, Delivery<View, T>>() { @Override public Delivery<View, T> call(View view) { return view == null ? null : new Delivery<>(view, notification); } }); } }) .filter(new Func1<Delivery<View, T>, Boolean>() { @Override public Boolean call(Delivery<View, T> delivery) { return delivery != null; } }) .take(1); } DeliverFirst(Observable<View> view); @Override Observable<Delivery<View, T>> call(Observable<T> observable); }
DeliverFirst implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { return observable.materialize() .take(1) .switchMap(new Func1<Notification<T>, Observable<? extends Delivery<View, T>>>() { @Override public Observable<? extends Delivery<View, T>> call(final Notification<T> notification) { return view.map(new Func1<View, Delivery<View, T>>() { @Override public Delivery<View, T> call(View view) { return view == null ? null : new Delivery<>(view, notification); } }); } }) .filter(new Func1<Delivery<View, T>, Boolean>() { @Override public Boolean call(Delivery<View, T> delivery) { return delivery != null; } }) .take(1); } DeliverFirst(Observable<View> view); @Override Observable<Delivery<View, T>> call(Observable<T> observable);  }