@Test public void testDroppedFailureIfOnFailureThrowAnExceptionWithoutSubscription() { AtomicReference<Throwable> captured = new AtomicReference<>(); Infrastructure.setDroppedExceptionHandler(captured::set); Subscriber<Integer> subscriber = Mocks.subscriber(2); doThrow(new IllegalArgumentException("boomB")).when(subscriber).onError(any(Throwable.class)); SafeSubscriber<Integer> safe = new SafeSubscriber<>(subscriber); safe.onError(new IOException("I/O")); assertThat(captured.get()) .isInstanceOf(CompositeException.class) .hasMessageContaining("I/O").hasMessageContaining("Subscription not set").hasMessageContaining("boomB"); }
@Override public void onError(Throwable t) { Objects.requireNonNull(t); if (done) { return; } done = true; if (upstream.get() == null) { Throwable npe = new NullPointerException("Subscription not set!"); try { downstream.onSubscribe(Subscriptions.empty()); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, e)); return; } try { downstream.onError(new CompositeException(t, npe)); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, npe, e)); } return; } try { downstream.onError(t); } catch (Throwable ex) { Infrastructure.handleDroppedException(new CompositeException(t, ex)); } }
SafeSubscriber implements Subscriber<T>, Subscription { @Override public void onError(Throwable t) { Objects.requireNonNull(t); if (done) { return; } done = true; if (upstream.get() == null) { Throwable npe = new NullPointerException("Subscription not set!"); try { downstream.onSubscribe(Subscriptions.empty()); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, e)); return; } try { downstream.onError(new CompositeException(t, npe)); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, npe, e)); } return; } try { downstream.onError(t); } catch (Throwable ex) { Infrastructure.handleDroppedException(new CompositeException(t, ex)); } } }
SafeSubscriber implements Subscriber<T>, Subscription { @Override public void onError(Throwable t) { Objects.requireNonNull(t); if (done) { return; } done = true; if (upstream.get() == null) { Throwable npe = new NullPointerException("Subscription not set!"); try { downstream.onSubscribe(Subscriptions.empty()); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, e)); return; } try { downstream.onError(new CompositeException(t, npe)); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, npe, e)); } return; } try { downstream.onError(t); } catch (Throwable ex) { Infrastructure.handleDroppedException(new CompositeException(t, ex)); } } SafeSubscriber(Subscriber<? super T> downstream); }
SafeSubscriber implements Subscriber<T>, Subscription { @Override public void onError(Throwable t) { Objects.requireNonNull(t); if (done) { return; } done = true; if (upstream.get() == null) { Throwable npe = new NullPointerException("Subscription not set!"); try { downstream.onSubscribe(Subscriptions.empty()); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, e)); return; } try { downstream.onError(new CompositeException(t, npe)); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, npe, e)); } return; } try { downstream.onError(t); } catch (Throwable ex) { Infrastructure.handleDroppedException(new CompositeException(t, ex)); } } SafeSubscriber(Subscriber<? super T> downstream); @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(); }
SafeSubscriber implements Subscriber<T>, Subscription { @Override public void onError(Throwable t) { Objects.requireNonNull(t); if (done) { return; } done = true; if (upstream.get() == null) { Throwable npe = new NullPointerException("Subscription not set!"); try { downstream.onSubscribe(Subscriptions.empty()); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, e)); return; } try { downstream.onError(new CompositeException(t, npe)); } catch (Throwable e) { Infrastructure.handleDroppedException(new CompositeException(t, npe, e)); } return; } try { downstream.onError(t); } catch (Throwable ex) { Infrastructure.handleDroppedException(new CompositeException(t, ex)); } } SafeSubscriber(Subscriber<? super T> downstream); @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();  }