SafeObserver implements Observer<T>, Disposable { @Override public void onError(@NonNull Throwable t) { if (done) { RxJavaPlugins.onError(t); return; } done = true; if (s == null) { Throwable npe = new NullPointerException("Subscription not set!"); try { actual.onSubscribe(EmptyDisposable.INSTANCE); } catch (Throwable e) { Exceptions.throwIfFatal(e); RxJavaPlugins.onError(new CompositeException(t, npe, e)); return; } try { actual.onError(new CompositeException(t, npe)); } catch (Throwable e) { Exceptions.throwIfFatal(e); RxJavaPlugins.onError(new CompositeException(t, npe, e)); } return; } if (t == null) { t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources."); } try { actual.onError(t); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); RxJavaPlugins.onError(new CompositeException(t, ex)); } } 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 onErrorFailure() { try { OBSERVER_ONERROR_FAIL().onError(new SafeObserverTestException("error!")); fail("expects exception to be thrown"); } catch (Exception e) { assertTrue(e instanceof SafeObserverTestException); assertEquals("onErrorFail", e.getMessage()); } }
@Test @Ignore("Observers can't throw") public void onErrorFailureSafe() { try { new SafeObserver<String>(OBSERVER_ONERROR_FAIL()).onError(new SafeObserverTestException("error!")); fail("expects exception to be thrown"); } catch (Exception e) { e.printStackTrace(); assertTrue(e instanceof RuntimeException); assertEquals("Error occurred when trying to propagate error to Observer.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 SafeObserverTestException); assertEquals("error!", e3.getMessage()); Throwable e4 = innerExceptions.get(1); assertTrue(e4 instanceof SafeObserverTestException); assertEquals("onErrorFail", e4.getMessage()); } }
@Test @Ignore("Observers can't throw") public void onErrorNotImplementedFailureSafe() { try { new SafeObserver<String>(OBSERVER_ONERROR_NOTIMPLEMENTED()).onError(new SafeObserverTestException("error!")); fail("expects exception to be thrown"); } catch (Exception e) { assertTrue(e.getCause() instanceof SafeObserverTestException); assertEquals("error!", e.getCause().getMessage()); } }
@Test public void onErrorWithoutOnSubscribe() { TestObserver<Integer> to = new TestObserver<Integer>(); SafeObserver<Integer> so = new SafeObserver<Integer>(to); so.onError(new TestException()); to.assertFailure(CompositeException.class); TestHelper.assertError(to, 0, TestException.class); TestHelper.assertError(to, 1, NullPointerException.class, "Subscription not set!"); }
@Test public void onErrorNoSubscribeCrash() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(true, 1, false, false, false); SafeObserver<Object> so = cd.toSafe(); so.onError(new TestException()); TestHelper.assertError(list, 0, CompositeException.class); List<Throwable> ce = TestHelper.compositeList(list.get(0)); TestHelper.assertError(ce, 0, TestException.class); TestHelper.assertError(ce, 1, NullPointerException.class, "Subscription not set!"); } finally { RxJavaPlugins.reset(); } }
@Test public void onErrorNoSubscribeOnErrorCrash() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(false, 1, true, false, false); SafeObserver<Object> so = cd.toSafe(); so.onError(new TestException()); TestHelper.assertError(list, 0, CompositeException.class); List<Throwable> ce = TestHelper.compositeList(list.get(0)); TestHelper.assertError(ce, 0, TestException.class); TestHelper.assertError(ce, 1, NullPointerException.class, "Subscription not set!"); TestHelper.assertError(ce, 2, TestException.class); } finally { RxJavaPlugins.reset(); } }