@Test public void testUnSubscribe() { TestObservable tA = new TestObservable(); TestObservable tB = new TestObservable(); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(Observable.create(tA), Observable.create(tB))); Subscription s = m.subscribe(stringObserver); tA.sendOnNext("Aone"); tB.sendOnNext("Bone"); s.unsubscribe(); tA.sendOnNext("Atwo"); tB.sendOnNext("Btwo"); tA.sendOnCompleted(); tB.sendOnCompleted(); verify(stringObserver, never()).onError(any(Throwable.class)); verify(stringObserver, times(1)).onNext("Aone"); verify(stringObserver, times(1)).onNext("Bone"); assertTrue(tA.unsubscribed); assertTrue(tB.unsubscribed); verify(stringObserver, never()).onNext("Atwo"); verify(stringObserver, never()).onNext("Btwo"); verify(stringObserver, never()).onCompleted(); }
public static <T> OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new MergeDelayErrorObservable<T>(sequences).onSubscribe(observer); } }; }
OperationMergeDelayError { public static <T> OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new MergeDelayErrorObservable<T>(sequences).onSubscribe(observer); } }; } }
OperationMergeDelayError { public static <T> OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new MergeDelayErrorObservable<T>(sequences).onSubscribe(observer); } }; }  }
OperationMergeDelayError { public static <T> OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new MergeDelayErrorObservable<T>(sequences).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences); static OnSubscribeFunc<T> mergeDelayError(final Observable<? extends T>... sequences); static OnSubscribeFunc<T> mergeDelayError(final List<? extends Observable<? extends T>> sequences); }
OperationMergeDelayError { public static <T> OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new MergeDelayErrorObservable<T>(sequences).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences); static OnSubscribeFunc<T> mergeDelayError(final Observable<? extends T>... sequences); static OnSubscribeFunc<T> mergeDelayError(final List<? extends Observable<? extends T>> sequences);  }