SafeObserver implements Observer<T>, Disposable { @Override public void onComplete() { if (done) { return; } done = true; if (s == null) { onCompleteNoSubscription(); return; } try { actual.onComplete(); } catch (Throwable e) { Exceptions.throwIfFatal(e); RxJavaPlugins.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 onCompleteFailure() { AtomicReference<Throwable> onError = new AtomicReference<Throwable>(); try { OBSERVER_ONCOMPLETED_FAIL(onError).onComplete(); fail("expects exception to be thrown"); } catch (Exception e) { assertNull(onError.get()); assertTrue(e instanceof SafeObserverTestException); assertEquals("onCompleteFail", e.getMessage()); } }
@Test @Ignore("Observers can't throw") public void testOnCompletedThrows() { final AtomicReference<Throwable> error = new AtomicReference<Throwable>(); SafeObserver<Integer> s = new SafeObserver<Integer>(new DefaultObserver<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable e) { error.set(e); } @Override public void onComplete() { throw new TestException(); } }); try { s.onComplete(); Assert.fail(); } catch (RuntimeException e) { assertNull(error.get()); } }
@Test public void onCompleteWithoutOnSubscribe() { TestObserver<Integer> to = new TestObserver<Integer>(); SafeObserver<Integer> so = new SafeObserver<Integer>(to); so.onComplete(); to.assertFailureAndMessage(NullPointerException.class, "Subscription not set!"); }
@Test public void onCompleteteNoSubscribeCrash() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(true, 1, false, true, false); SafeObserver<Object> so = cd.toSafe(); so.onComplete(); 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 onCompleteteNoSubscribeOnErrorCrash() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(false, 1, true, true, false); SafeObserver<Object> so = cd.toSafe(); so.onComplete(); 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); } finally { RxJavaPlugins.reset(); } }