@Test public void testConcurrency() { ExecutorService executor = Executors.newFixedThreadPool(20); try { ConcurrentSubscriber sub = new ConcurrentSubscriber(); Subscriber<String> serialized = serialized(new SafeSubscriber<>(sub)); serialized.onSubscribe(mock(Subscription.class)); Future<?> f1 = executor.submit(new OnNextThread(serialized, 12000)); Future<?> f2 = executor.submit(new OnNextThread(serialized, 5000)); Future<?> f3 = executor.submit(new OnNextThread(serialized, 75000)); Future<?> f4 = executor.submit(new OnNextThread(serialized, 13500)); Future<?> f5 = executor.submit(new OnNextThread(serialized, 22000)); Future<?> f6 = executor.submit(new OnNextThread(serialized, 15000)); Future<?> f7 = executor.submit(new OnNextThread(serialized, 7500)); Future<?> f8 = executor.submit(new OnNextThread(serialized, 23500)); Future<?> f10 = executor .submit(new CompletionThread(serialized, TestConcurrencySubscriberEvent.onComplete, 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 = sub.assertEvents(null); assertThat(numNextEvents).isEqualTo(12000 + 5000 + 75000 + 13500 + 22000 + 15000 + 7500 + 23500); } finally { executor.shutdownNow(); } }
@Override public void onSubscribe(Subscription s) { if (upstream.compareAndSet(null, s)) { downstream.onSubscribe(this); } else { s.cancel(); Infrastructure.handleDroppedException(new IllegalStateException("Subscription already set")); } }
SerializedSubscriber implements Subscription, MultiSubscriber<T> { @Override public void onSubscribe(Subscription s) { if (upstream.compareAndSet(null, s)) { downstream.onSubscribe(this); } else { s.cancel(); Infrastructure.handleDroppedException(new IllegalStateException("Subscription already set")); } } }
SerializedSubscriber implements Subscription, MultiSubscriber<T> { @Override public void onSubscribe(Subscription s) { if (upstream.compareAndSet(null, s)) { downstream.onSubscribe(this); } else { s.cancel(); Infrastructure.handleDroppedException(new IllegalStateException("Subscription already set")); } } SerializedSubscriber(Subscriber<? super T> downstream); }
SerializedSubscriber implements Subscription, MultiSubscriber<T> { @Override public void onSubscribe(Subscription s) { if (upstream.compareAndSet(null, s)) { downstream.onSubscribe(this); } else { s.cancel(); Infrastructure.handleDroppedException(new IllegalStateException("Subscription already set")); } } SerializedSubscriber(Subscriber<? super T> downstream); @Override void onSubscribe(Subscription s); @Override void onItem(T t); @Override void onFailure(Throwable t); @Override void onCompletion(); @Override void request(long n); @Override void cancel(); }
SerializedSubscriber implements Subscription, MultiSubscriber<T> { @Override public void onSubscribe(Subscription s) { if (upstream.compareAndSet(null, s)) { downstream.onSubscribe(this); } else { s.cancel(); Infrastructure.handleDroppedException(new IllegalStateException("Subscription already set")); } } SerializedSubscriber(Subscriber<? super T> downstream); @Override void onSubscribe(Subscription s); @Override void onItem(T t); @Override void onFailure(Throwable t); @Override void onCompletion(); @Override void request(long n); @Override void cancel();  }