OperationSynchronize { public static <T> OnSubscribeFunc<T> synchronize(Observable<? extends T> observable) { return new Synchronize<T>(observable, null); }  static OnSubscribeFunc<T> synchronize(Observable<? extends T> observable); static OnSubscribeFunc<T> synchronize(Observable<? extends T> observable, Object lock);  }
@Test public void testOnCompletedAfterUnSubscribe() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); Subscription ws = st.subscribe(w); System.out.println("ws: " + ws); t.sendOnNext("one"); ws.unsubscribe(); System.out.println("send onCompleted"); t.sendOnCompleted(); verify(w, times(1)).onNext("one"); verify(w, Mockito.never()).onCompleted(); }
@Test public void testOnNextAfterUnSubscribe() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); Subscription ws = st.subscribe(w); t.sendOnNext("one"); ws.unsubscribe(); t.sendOnNext("two"); verify(w, times(1)).onNext("one"); verify(w, Mockito.never()).onNext("two"); }
@Test public void testOnErrorAfterUnSubscribe() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); Subscription ws = st.subscribe(w); t.sendOnNext("one"); ws.unsubscribe(); t.sendOnError(new RuntimeException("bad")); verify(w, times(1)).onNext("one"); verify(w, Mockito.never()).onError(any(Throwable.class)); }
@Test public void testOnNextAfterOnError() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); @SuppressWarnings("unused") Subscription ws = st.subscribe(w); t.sendOnNext("one"); t.sendOnError(new RuntimeException("bad")); t.sendOnNext("two"); verify(w, times(1)).onNext("one"); verify(w, times(1)).onError(any(Throwable.class)); verify(w, Mockito.never()).onNext("two"); }
@Test public void testOnCompletedAfterOnError() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); @SuppressWarnings("unused") Subscription ws = st.subscribe(w); t.sendOnNext("one"); t.sendOnError(new RuntimeException("bad")); t.sendOnCompleted(); verify(w, times(1)).onNext("one"); verify(w, times(1)).onError(any(Throwable.class)); verify(w, Mockito.never()).onCompleted(); }
@Test public void testOnNextAfterOnCompleted() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); @SuppressWarnings("unused") Subscription ws = st.subscribe(w); t.sendOnNext("one"); t.sendOnCompleted(); t.sendOnNext("two"); verify(w, times(1)).onNext("one"); verify(w, Mockito.never()).onNext("two"); verify(w, times(1)).onCompleted(); verify(w, Mockito.never()).onError(any(Throwable.class)); }
@Test public void testOnErrorAfterOnCompleted() { TestObservable t = new TestObservable(null); Observable<String> st = Observable.create(synchronize(Observable.create(t))); @SuppressWarnings("unchecked") Observer<String> w = mock(Observer.class); @SuppressWarnings("unused") Subscription ws = st.subscribe(w); t.sendOnNext("one"); t.sendOnCompleted(); t.sendOnError(new RuntimeException("bad")); verify(w, times(1)).onNext("one"); verify(w, times(1)).onCompleted(); verify(w, Mockito.never()).onError(any(Throwable.class)); }