@Test public void testMultiThreadedBasic() { TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three"); Observable<String> w = Observable.create(onSubscribe); BusyObserver busyObserver = new BusyObserver(); Observer<String> aw = serializedObserver(busyObserver); w.subscribe(aw); onSubscribe.waitToFinish(); assertEquals(3, busyObserver.onNextCount.get()); assertFalse(busyObserver.onError); assertTrue(busyObserver.onCompleted); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busyObserver.maxConcurrentThreads.get()); }
public SerializedObserver(Observer<? super T> s) { this.actual = s; }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } SerializedObserver(Observer<? super T> s); }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } SerializedObserver(Observer<? super T> s); @Override void onCompleted(); @Override void onError(final Throwable e); @Override void onNext(T t); }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } SerializedObserver(Observer<? super T> s); @Override void onCompleted(); @Override void onError(final Throwable e); @Override void onNext(T t);  }