@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(); } } }
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);  }