SafeSubscriber implements FlowableSubscriber<T>, Subscription { @Override public void onSubscribe(Subscription s) { if (SubscriptionHelper.validate(this.s, s)) { this.s = s; try { actual.onSubscribe(this); } catch (Throwable e) { Exceptions.throwIfFatal(e); done = true; try { s.cancel(); } catch (Throwable e1) { Exceptions.throwIfFatal(e1); RxJavaPlugins.onError(new CompositeException(e, e1)); return; } RxJavaPlugins.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 dispose() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); SafeSubscriber<Integer> so = new SafeSubscriber<Integer>(ts); BooleanSubscription d = new BooleanSubscription(); so.onSubscribe(d); ts.dispose(); assertTrue(d.isCancelled()); }
@Test public void onSubscribeTwice() { List<Throwable> list = TestHelper.trackPluginErrors(); try { CrashDummy cd = new CrashDummy(false, 1, false, false, false, false); SafeSubscriber<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, false); SafeSubscriber<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, false); SafeSubscriber<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, "cancel()"); } finally { RxJavaPlugins.reset(); } }