OperationDematerialize { public static <T> OnSubscribeFunc<T> dematerialize(final Observable<? extends Notification<? extends T>> sequence) { return new DematerializeObservable<T>(sequence); }  static OnSubscribeFunc<T> dematerialize(final Observable<? extends Notification<? extends T>> sequence);  }
@Test @SuppressWarnings("unchecked") public void testDematerialize1() { Observable<Notification<Integer>> notifications = Observable.from(1, 2).materialize(); Observable<Integer> dematerialize = notifications.dematerialize(); Observer<Integer> observer = mock(Observer.class); dematerialize.subscribe(observer); verify(observer, times(1)).onNext(1); verify(observer, times(1)).onNext(2); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test @SuppressWarnings("unchecked") public void testDematerialize2() { Throwable exception = new Throwable("test"); Observable<Integer> observable = Observable.error(exception); Observable<Integer> dematerialize = Observable.create(dematerialize(observable.materialize())); Observer<Integer> observer = mock(Observer.class); dematerialize.subscribe(observer); verify(observer, times(1)).onError(exception); verify(observer, times(0)).onCompleted(); verify(observer, times(0)).onNext(any(Integer.class)); }
@Test @SuppressWarnings("unchecked") public void testDematerialize3() { Exception exception = new Exception("test"); Observable<Integer> observable = Observable.error(exception); Observable<Integer> dematerialize = Observable.create(dematerialize(observable.materialize())); Observer<Integer> observer = mock(Observer.class); dematerialize.subscribe(observer); verify(observer, times(1)).onError(exception); verify(observer, times(0)).onCompleted(); verify(observer, times(0)).onNext(any(Integer.class)); }