@Test(timeout = 60000) public void testBufferCompletion() throws Exception { rule.subject.onCompleted(); rule.subscribeAndAssertValues(rule.subject); }
@Override public void onCompleted() { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = null; } } if (null != p) { p.terminateAndDrain(); } }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onCompleted() { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = null; } } if (null != p) { p.terminateAndDrain(); } } }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onCompleted() { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = null; } } if (null != p) { p.terminateAndDrain(); } } protected  UnicastBufferingSubject(OnSubscribe<T> onSubscribe, State<T> state); }
UnicastBufferingSubject extends Subject<T, T> { @Override public void onCompleted() { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = null; } } if (null != p) { p.terminateAndDrain(); } } 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 onCompleted() { BackpressureDrainManager p = null; synchronized (state) { if (null != state.producer) { p = state.producer; } else { state.terminatedBeforeSubscribe = true; state.errorBeforeSubscribe = null; } } if (null != p) { p.terminateAndDrain(); } } 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);  }