@Test(timeout = 60000) public void testBufferError() throws Exception { rule.subject.onError(new IllegalStateException()); TestSubscriber<String> sub = new TestSubscriber<>(); rule.subject.subscribe(sub); sub.assertTerminalEvent(); sub.assertError(IllegalStateException.class); }
@Override public void onError(Throwable e) { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = e; } } if (null != p) { p.terminateAndDrain(e); } }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onError(Throwable e) { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = e; } } if (null != p) { p.terminateAndDrain(e); } } }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onError(Throwable e) { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = e; } } if (null != p) { p.terminateAndDrain(e); } } protected  UnicastBufferingSubject(OnSubscribe<T> onSubscribe, State<T> state); }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onError(Throwable e) { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = e; } } if (null != p) { p.terminateAndDrain(e); } } protected  UnicastBufferingSubject(OnSubscribe<T> onSubscribe, State<T> state); static UnicastBufferingSubject<T> create(long bufferSize); boolean isTerminated(); @Override boolean hasObservers(); @Override void onCompleted(); @Override void onError(Throwable e); @Override void onNext(T t); boolean offerNext(T next); }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onError(Throwable e) { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = e; } } if (null != p) { p.terminateAndDrain(e); } } protected  UnicastBufferingSubject(OnSubscribe<T> onSubscribe, State<T> state); static UnicastBufferingSubject<T> create(long bufferSize); boolean isTerminated(); @Override boolean hasObservers(); @Override void onCompleted(); @Override void onError(Throwable e); @Override void onNext(T t); boolean offerNext(T next);  }