SerializedSubscriber implements FlowableSubscriber<T>, Subscription { public SerializedSubscriber(Subscriber<? super T> actual) { this(actual, false); } SerializedSubscriber(Subscriber<? super T> actual);  SerializedSubscriber(Subscriber<? super T> actual, boolean delayError); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void request(long n); @Override void cancel();  }
@Test public void testMultiThreadedBasic() { TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three"); Flowable<String> w = Flowable.unsafeCreate(onSubscribe); BusySubscriber busySubscriber = new BusySubscriber(); Subscriber<String> aw = serializedSubscriber(busySubscriber); w.subscribe(aw); onSubscribe.waitToFinish(); assertEquals(3, busySubscriber.onNextCount.get()); assertFalse(busySubscriber.onError); assertTrue(busySubscriber.onComplete); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busySubscriber.maxConcurrentThreads.get()); }
@Test(timeout = 1000) public void testMultiThreadedWithNPE() throws InterruptedException { TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three", null); Flowable<String> w = Flowable.unsafeCreate(onSubscribe); BusySubscriber busySubscriber = new BusySubscriber(); Subscriber<String> aw = serializedSubscriber(busySubscriber); w.subscribe(aw); onSubscribe.waitToFinish(); busySubscriber.terminalEvent.await(); System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + " Subscriber maxConcurrentThreads: " + busySubscriber.maxConcurrentThreads.get()); assertTrue(busySubscriber.onNextCount.get() < 4); assertTrue(busySubscriber.onError); assertFalse(busySubscriber.onComplete); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busySubscriber.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"); Flowable<String> w = Flowable.unsafeCreate(onSubscribe); BusySubscriber busySubscriber = new BusySubscriber(); Subscriber<String> aw = serializedSubscriber(busySubscriber); w.subscribe(aw); onSubscribe.waitToFinish(); System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + " Subscriber maxConcurrentThreads: " + busySubscriber.maxConcurrentThreads.get()); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busySubscriber.maxConcurrentThreads.get()); System.out.println("onNext count: " + busySubscriber.onNextCount.get()); assertFalse(busySubscriber.onComplete); assertTrue(busySubscriber.onError); assertTrue(busySubscriber.onNextCount.get() < 9); } }
@Test public void runOutOfOrderConcurrencyTest() { ExecutorService tp = Executors.newFixedThreadPool(20); try { TestConcurrencySubscriber tw = new TestConcurrencySubscriber(); Subscriber<String> w = serializedSubscriber(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, TestConcurrencySubscriberEvent.onComplete, f1, f2, f3, f4)); try { Thread.sleep(1); } catch (InterruptedException e) { } Future<?> f11 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onComplete, f4, f6, f7)); Future<?> f12 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onComplete, f4, f6, f7)); Future<?> f13 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onComplete, f4, f6, f7)); Future<?> f14 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onComplete, f4, f6, f7)); Future<?> f15 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onError, f1, f2, f3, f4)); Future<?> f16 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onError, f1, f2, f3, f4)); Future<?> f17 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.onError, f1, f2, f3, f4)); Future<?> f18 = tp.submit(new CompletionThread(w, TestConcurrencySubscriberEvent.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(); } } }