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