@Test public void testMultiThreadedWithNPEinMiddle() { int n = 10; for (int i = 0; i < n; i++) { TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three", null, "four", "five", "six", "seven", "eight", "nine"); Observable<String> w = Observable.create(onSubscribe); BusyObserver busyObserver = new BusyObserver(); Observer<String> aw = serializedObserver(busyObserver); w.subscribe(aw); onSubscribe.waitToFinish(); System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + " Observer maxConcurrentThreads: " + busyObserver.maxConcurrentThreads.get()); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busyObserver.maxConcurrentThreads.get()); System.out.println("onNext count: " + busyObserver.onNextCount.get()); assertFalse(busyObserver.onCompleted); assertTrue(busyObserver.onError); assertTrue(busyObserver.onNextCount.get() < 9); } }
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);  }