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);  }
@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()); }
@Test(timeout = 1000) public void testMultiThreadedWithNPE() throws InterruptedException { TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three", null); Observable<String> w = Observable.create(onSubscribe); BusyObserver busyObserver = new BusyObserver(); Observer<String> aw = serializedObserver(busyObserver); w.subscribe(aw); onSubscribe.waitToFinish(); busyObserver.terminalEvent.await(); System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + " Observer maxConcurrentThreads: " + busyObserver.maxConcurrentThreads.get()); assertTrue(busyObserver.onNextCount.get() < 4); assertTrue(busyObserver.onError); assertFalse(busyObserver.onCompleted); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busyObserver.maxConcurrentThreads.get()); }
@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); } }
@Test public void runOutOfOrderConcurrencyTest() { ExecutorService tp = Executors.newFixedThreadPool(20); try { TestConcurrencyObserver tw = new TestConcurrencyObserver(); Observer<String> w = serializedObserver(new SafeSubscriber<String>(tw)); Future<?> f1 = tp.submit(new OnNextThread(w, 12000)); Future<?> f2 = tp.submit(new OnNextThread(w, 5000)); Future<?> f3 = tp.submit(new OnNextThread(w, 75000)); Future<?> f4 = tp.submit(new OnNextThread(w, 13500)); Future<?> f5 = tp.submit(new OnNextThread(w, 22000)); Future<?> f6 = tp.submit(new OnNextThread(w, 15000)); Future<?> f7 = tp.submit(new OnNextThread(w, 7500)); Future<?> f8 = tp.submit(new OnNextThread(w, 23500)); Future<?> f10 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onCompleted, f1, f2, f3, f4)); try { Thread.sleep(1); } catch (InterruptedException e) { } Future<?> f11 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onCompleted, f4, f6, f7)); Future<?> f12 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onCompleted, f4, f6, f7)); Future<?> f13 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onCompleted, f4, f6, f7)); Future<?> f14 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onCompleted, f4, f6, f7)); Future<?> f15 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onError, f1, f2, f3, f4)); Future<?> f16 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onError, f1, f2, f3, f4)); Future<?> f17 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onError, f1, f2, f3, f4)); Future<?> f18 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onError, f1, f2, f3, f4)); waitOnThreads(f1, f2, f3, f4, f5, f6, f7, f8, f10, f11, f12, f13, f14, f15, f16, f17, f18); @SuppressWarnings("unused") int numNextEvents = tw.assertEvents(null); } catch (Throwable e) { fail("Concurrency test failed: " + e.getMessage()); e.printStackTrace(); } finally { tp.shutdown(); try { tp.awaitTermination(5000, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { e.printStackTrace(); } } }
@Test public void runConcurrencyTest() { ExecutorService tp = Executors.newFixedThreadPool(20); try { TestConcurrencyObserver tw = new TestConcurrencyObserver(); Observer<String> w = serializedObserver(new SafeSubscriber<String>(tw)); Future<?> f1 = tp.submit(new OnNextThread(w, 12000)); Future<?> f2 = tp.submit(new OnNextThread(w, 5000)); Future<?> f3 = tp.submit(new OnNextThread(w, 75000)); Future<?> f4 = tp.submit(new OnNextThread(w, 13500)); Future<?> f5 = tp.submit(new OnNextThread(w, 22000)); Future<?> f6 = tp.submit(new OnNextThread(w, 15000)); Future<?> f7 = tp.submit(new OnNextThread(w, 7500)); Future<?> f8 = tp.submit(new OnNextThread(w, 23500)); Future<?> f10 = tp.submit(new CompletionThread(w, TestConcurrencyObserverEvent.onCompleted, f1, f2, f3, f4, f5, f6, f7, f8)); try { Thread.sleep(1); } catch (InterruptedException e) { } waitOnThreads(f1, f2, f3, f4, f5, f6, f7, f8, f10); int numNextEvents = tw.assertEvents(null); assertEquals(173500, numNextEvents); } catch (Throwable e) { fail("Concurrency test failed: " + e.getMessage()); e.printStackTrace(); } finally { tp.shutdown(); try { tp.awaitTermination(25000, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { e.printStackTrace(); } } }
@Ignore @Test public void testThreadStarvation() throws InterruptedException { TestSubscriber<String> to = new TestSubscriber<String>(new Observer<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String t) { try { Thread.sleep(1); } catch (InterruptedException e) { } } }); Observer<String> o = serializedObserver(to); AtomicInteger p1 = new AtomicInteger(); AtomicInteger p2 = new AtomicInteger(); Subscription s1 = infinite(p1).subscribe(o); Subscription s2 = infinite(p2).subscribe(o); Thread.sleep(100); System.out.println("p1: " + p1.get() + " p2: " + p2.get() + " => should be close to each other unless we have thread starvation"); assertEquals(p1.get(), p2.get(), 10000); s1.unsubscribe(); s2.unsubscribe(); }