@Test public void onNextFailure() { AtomicReference<Throwable> onError = new AtomicReference<Throwable>(); try { OBSERVER_ONNEXT_FAIL(onError).onNext("one"); fail("expects exception to be thrown"); } catch (Exception e) { assertNull(onError.get()); assertTrue(e instanceof SafeSubscriberTestException); assertEquals("onNextFail", e.getMessage()); } }
@Override public void onNext(T t) { if (done) { return; } if (s == null) { onNextNoSubscription(); return; } if (t == null) { Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(ex, e1)); return; } onError(ex); return; } try { actual.onNext(t); } catch (Throwable e) { Exceptions.throwIfFatal(e); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(e, e1)); return; } onError(e); } }
SafeSubscriber implements FlowableSubscriber<T>, Subscription { @Override public void onNext(T t) { if (done) { return; } if (s == null) { onNextNoSubscription(); return; } if (t == null) { Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(ex, e1)); return; } onError(ex); return; } try { actual.onNext(t); } catch (Throwable e) { Exceptions.throwIfFatal(e); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(e, e1)); return; } onError(e); } } }
SafeSubscriber implements FlowableSubscriber<T>, Subscription { @Override public void onNext(T t) { if (done) { return; } if (s == null) { onNextNoSubscription(); return; } if (t == null) { Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(ex, e1)); return; } onError(ex); return; } try { actual.onNext(t); } catch (Throwable e) { Exceptions.throwIfFatal(e); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(e, e1)); return; } onError(e); } } SafeSubscriber(Subscriber<? super T> actual); }
SafeSubscriber implements FlowableSubscriber<T>, Subscription { @Override public void onNext(T t) { if (done) { return; } if (s == null) { onNextNoSubscription(); return; } if (t == null) { Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(ex, e1)); return; } onError(ex); return; } try { actual.onNext(t); } catch (Throwable e) { Exceptions.throwIfFatal(e); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(e, e1)); return; } onError(e); } } SafeSubscriber(Subscriber<? super T> actual); @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 FlowableSubscriber<T>, Subscription { @Override public void onNext(T t) { if (done) { return; } if (s == null) { onNextNoSubscription(); return; } if (t == null) { Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(ex, e1)); return; } onError(ex); return; } try { actual.onNext(t); } catch (Throwable e) { Exceptions.throwIfFatal(e); try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(e, e1)); return; } onError(e); } } SafeSubscriber(Subscriber<? super T> actual); @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();  }