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();  }
@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()); } }
@Test public void onNextOnErrorFailure() { try { OBSERVER_ONNEXT_ONERROR_FAIL().onNext("one"); fail("expects exception to be thrown"); } catch (Exception e) { e.printStackTrace(); assertTrue(e instanceof SafeSubscriberTestException); assertEquals("onNextFail", e.getMessage()); } }
@Test @Ignore("Observers can't throw") public void onNextOnErrorFailureSafe() { try { new SafeSubscriber<String>(OBSERVER_ONNEXT_ONERROR_FAIL()).onNext("one"); fail("expects exception to be thrown"); } catch (Exception e) { e.printStackTrace(); assertTrue(e instanceof RuntimeException); assertEquals("Error occurred when trying to propagate error to Subscriber.onError", e.getMessage()); Throwable e2 = e.getCause(); assertTrue(e2 instanceof CompositeException); List<Throwable> innerExceptions = ((CompositeException) e2).getExceptions(); assertEquals(2, innerExceptions.size()); Throwable e3 = innerExceptions.get(0); assertTrue(e3 instanceof SafeSubscriberTestException); assertEquals("onNextFail", e3.getMessage()); Throwable e4 = innerExceptions.get(1); assertTrue(e4 instanceof SafeSubscriberTestException); assertEquals("onErrorFail", e4.getMessage()); } }
@Test public void onNextWithoutOnSubscribe() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); SafeSubscriber<Integer> so = new SafeSubscriber<Integer>(ts); so.onNext(1); ts.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, false); SafeSubscriber<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, false); SafeSubscriber<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(); } }