@Test public void testReplay() 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(); DeliverReplay<Object, Integer> restartable = new DeliverReplay<>(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))); } } ); } }); subject.onNext(1); subject.onNext(2); subject.onNext(3); testSubscriber.assertNotCompleted(); testSubscriber.assertNoValues(); view.onNext(100); deliveries.add(new Delivery<Object, Integer>(100, Notification.createOnNext(1))); deliveries.add(new Delivery<Object, Integer>(100, Notification.createOnNext(2))); deliveries.add(new Delivery<Object, Integer>(100, Notification.createOnNext(3))); testSubscriber.assertValueCount(3); testSubscriber.assertNotCompleted(); view.onNext(null); testSubscriber.assertValueCount(3); testSubscriber.assertNotCompleted(); view.onNext(101); deliveries.add(new Delivery<Object, Integer>(101, Notification.createOnNext(1))); deliveries.add(new Delivery<Object, Integer>(101, Notification.createOnNext(2))); deliveries.add(new Delivery<Object, Integer>(101, Notification.createOnNext(3))); testSubscriber.assertValueCount(6); testSubscriber.assertNotCompleted(); Throwable throwable = new Throwable(); subject.onError(throwable); deliveries.add(new Delivery<Object, Integer>(101, Notification.<Integer>createOnError(throwable))); testSubscriber.assertValueCount(7); testSubscriber.assertNotCompleted(); testSubscriber.assertReceivedOnNext(deliveries); subscription.unsubscribe(); assertFalse(subject.hasObservers()); assertFalse(view.hasObservers()); }
@Override public Observable<Delivery<View, T>> call(Observable<T> observable) { final ReplaySubject<Notification<T>> subject = ReplaySubject.create(); final Subscription subscription = observable .materialize() .filter(new Func1<Notification<T>, Boolean>() { @Override public Boolean call(Notification<T> notification) { return !notification.isOnCompleted(); } }) .subscribe(subject); return view .switchMap(new Func1<View, Observable<Delivery<View, T>>>() { @Override public Observable<Delivery<View, T>> call(final View view) { return view == null ? Observable.<Delivery<View, T>>never() : subject .map(new Func1<Notification<T>, Delivery<View, T>>() { @Override public Delivery<View, T> call(Notification<T> notification) { return new Delivery<>(view, notification); } }); } }) .doOnUnsubscribe(new Action0() { @Override public void call() { subscription.unsubscribe(); } }); }
DeliverReplay implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { final ReplaySubject<Notification<T>> subject = ReplaySubject.create(); final Subscription subscription = observable .materialize() .filter(new Func1<Notification<T>, Boolean>() { @Override public Boolean call(Notification<T> notification) { return !notification.isOnCompleted(); } }) .subscribe(subject); return view .switchMap(new Func1<View, Observable<Delivery<View, T>>>() { @Override public Observable<Delivery<View, T>> call(final View view) { return view == null ? Observable.<Delivery<View, T>>never() : subject .map(new Func1<Notification<T>, Delivery<View, T>>() { @Override public Delivery<View, T> call(Notification<T> notification) { return new Delivery<>(view, notification); } }); } }) .doOnUnsubscribe(new Action0() { @Override public void call() { subscription.unsubscribe(); } }); } }
DeliverReplay implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { final ReplaySubject<Notification<T>> subject = ReplaySubject.create(); final Subscription subscription = observable .materialize() .filter(new Func1<Notification<T>, Boolean>() { @Override public Boolean call(Notification<T> notification) { return !notification.isOnCompleted(); } }) .subscribe(subject); return view .switchMap(new Func1<View, Observable<Delivery<View, T>>>() { @Override public Observable<Delivery<View, T>> call(final View view) { return view == null ? Observable.<Delivery<View, T>>never() : subject .map(new Func1<Notification<T>, Delivery<View, T>>() { @Override public Delivery<View, T> call(Notification<T> notification) { return new Delivery<>(view, notification); } }); } }) .doOnUnsubscribe(new Action0() { @Override public void call() { subscription.unsubscribe(); } }); } DeliverReplay(Observable<View> view); }
DeliverReplay implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { final ReplaySubject<Notification<T>> subject = ReplaySubject.create(); final Subscription subscription = observable .materialize() .filter(new Func1<Notification<T>, Boolean>() { @Override public Boolean call(Notification<T> notification) { return !notification.isOnCompleted(); } }) .subscribe(subject); return view .switchMap(new Func1<View, Observable<Delivery<View, T>>>() { @Override public Observable<Delivery<View, T>> call(final View view) { return view == null ? Observable.<Delivery<View, T>>never() : subject .map(new Func1<Notification<T>, Delivery<View, T>>() { @Override public Delivery<View, T> call(Notification<T> notification) { return new Delivery<>(view, notification); } }); } }) .doOnUnsubscribe(new Action0() { @Override public void call() { subscription.unsubscribe(); } }); } DeliverReplay(Observable<View> view); @Override Observable<Delivery<View, T>> call(Observable<T> observable); }
DeliverReplay implements Observable.Transformer<T, Delivery<View, T>> { @Override public Observable<Delivery<View, T>> call(Observable<T> observable) { final ReplaySubject<Notification<T>> subject = ReplaySubject.create(); final Subscription subscription = observable .materialize() .filter(new Func1<Notification<T>, Boolean>() { @Override public Boolean call(Notification<T> notification) { return !notification.isOnCompleted(); } }) .subscribe(subject); return view .switchMap(new Func1<View, Observable<Delivery<View, T>>>() { @Override public Observable<Delivery<View, T>> call(final View view) { return view == null ? Observable.<Delivery<View, T>>never() : subject .map(new Func1<Notification<T>, Delivery<View, T>>() { @Override public Delivery<View, T> call(Notification<T> notification) { return new Delivery<>(view, notification); } }); } }) .doOnUnsubscribe(new Action0() { @Override public void call() { subscription.unsubscribe(); } }); } DeliverReplay(Observable<View> view); @Override Observable<Delivery<View, T>> call(Observable<T> observable);  }