SafeObserver implements Observer<T>, Disposable { @Override public void onSubscribe(@NonNull Disposable s) { if (DisposableHelper.validate(this.s, s)) { this.s = s; try { actual.onSubscribe(this); } catch (Throwable e) { Exceptions.throwIfFatal(e); done = true; try { s.dispose(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); RxJavaPlugins.onError(new CompositeException(e, e1)); return; } 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 onSubscribeTwice() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(false, 1, false, false, false); SafeObserver<Object> so = cd.toSafe(); so.onSubscribe(cd); so.onSubscribe(cd); TestHelper.assertError(list, 0, IllegalStateException.class); } finally { RxJavaPlugins.reset(); } }
@Test public void onSubscribeCrashes() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(true, 1, false, false, false); SafeObserver<Object> so = cd.toSafe(); so.onSubscribe(cd); TestHelper.assertUndeliverable(list, 0, TestException.class, "onSubscribe()"); } finally { RxJavaPlugins.reset(); } }
@Test public void onSubscribeAndDisposeCrashes() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(true, 1, false, false, true); SafeObserver<Object> so = cd.toSafe(); so.onSubscribe(cd); TestHelper.assertError(list, 0, CompositeException.class); List<Throwable> ce = TestHelper.compositeList(list.get(0)); TestHelper.assertError(ce, 0, TestException.class, "onSubscribe()"); TestHelper.assertError(ce, 1, TestException.class, "dispose()"); } finally { RxJavaPlugins.reset(); } }