@Test public void testOnceThrowable() throws Exception { PublishSubject<Object> view = PublishSubject.create(); final TestSubscriber<Delivery<Object, Integer>> testSubscriber = new TestSubscriber<>(); final ArrayList<Delivery<Object, Integer>> deliveries = new ArrayList<>(); final PublishSubject<Integer> subject = PublishSubject.create(); DeliverFirst<Object, Integer> restartable = new DeliverFirst<>(view); Subscription subscription = restartable.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) { testSubscriber.onNext(new Delivery<>(o, Notification.createOnNext(integer))); } }, new Action2<Object, Throwable>() { @Override public void call(Object o, Throwable throwable) { testSubscriber.onNext(new Delivery<>(o, Notification.<Integer>createOnError(throwable))); } } ); } }); Throwable throwable = new Throwable(); subject.onError(throwable); testSubscriber.assertNotCompleted(); testSubscriber.assertNoValues(); view.onNext(100); deliveries.add(new Delivery<Object, Integer>(100, Notification.<Integer>createOnError(throwable))); testSubscriber.assertValueCount(1); testSubscriber.assertReceivedOnNext(deliveries); subscription.unsubscribe(); assertFalse(subject.hasObservers()); assertFalse(view.hasObservers()); }
@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);  }