SafeObserver implements Observer<T>, Disposable { @Override public void onNext(@NonNull 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.dispose(); } 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.dispose(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); onError(new CompositeException(e, e1)); return; } onError(e); } } SafeObserver(@NonNull Observer<? super T> actual); @Override void onSubscribe(@NonNull Disposable s); @Override void dispose(); @Override boolean isDisposed(); @Override void onNext(@NonNull T t); @Override void onError(@NonNull Throwable t); @Override void onComplete();  }
@Test public void onNextWithoutOnSubscribe() { TestObserver<Integer> to = new TestObserver<Integer>(); SafeObserver<Integer> so = new SafeObserver<Integer>(to); so.onNext(1); to.assertFailureAndMessage(NullPointerException.class, "Subscription not set!"); }
@Test public void onNextOnSubscribeCrash() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(true, 1, false, false, false); SafeObserver<Object> so = cd.toSafe(); so.onNext(1); TestHelper.assertError(list, 0, CompositeException.class); List<Throwable> ce = TestHelper.compositeList(list.get(0)); TestHelper.assertError(ce, 0, NullPointerException.class, "Subscription not set!"); TestHelper.assertError(ce, 1, TestException.class, "onSubscribe()"); } finally { RxJavaPlugins.reset(); } }
@Test public void noSubscribeOnErrorCrashes() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(false, 1, true, false, false); SafeObserver<Object> so = cd.toSafe(); so.onNext(1); TestHelper.assertError(list, 0, CompositeException.class); List<Throwable> ce = TestHelper.compositeList(list.get(0)); TestHelper.assertError(ce, 0, NullPointerException.class, "Subscription not set!"); TestHelper.assertError(ce, 1, TestException.class, "onError(java.lang.NullPointerException: Subscription not set!)"); } finally { RxJavaPlugins.reset(); } }