### Question:
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();  }### Answer:
@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(); } }
@Test public void testOnFailureReentry() { List<Throwable> failures = new CopyOnWriteArrayList<>(); Infrastructure.setDroppedExceptionHandler(failures::add); final AtomicReference<Subscriber<Integer>> reference = new AtomicReference<>(); AssertSubscriber<Integer> subscriber = new AssertSubscriber<Integer>(Long.MAX_VALUE) { @Override public void onNext(Integer v) { reference.get().onError(new TestException("boom-1")); reference.get().onError(new TestException("boom-2")); super.onNext(v); } }; SerializedSubscriber<Integer> serialized = new SerializedSubscriber<>(subscriber); serialized.onSubscribe(mock(Subscription.class)); reference.set(serialized); serialized.onNext(1); subscriber .assertReceived(1) .assertHasFailedWith(TestException.class, "boom-1"); assertThat(failures).hasSize(1).allSatisfy(f -> assertThat(f).isInstanceOf(TestException.class) .hasMessageContaining("boom-2")); }
@Test public void testOnCompleteReentry() { final AtomicReference<Subscriber<Integer>> reference = new AtomicReference<>(); AssertSubscriber<Integer> subscriber = new AssertSubscriber<Integer>(Long.MAX_VALUE) { @Override public void onNext(Integer v) { reference.get().onComplete(); reference.get().onComplete(); super.onNext(v); } }; SerializedSubscriber<Integer> serialized = new SerializedSubscriber<>(subscriber); serialized.onSubscribe(mock(Subscription.class)); reference.set(serialized); subscriber.onNext(1); subscriber .assertReceived(1) .assertCompletedSuccessfully(); }