@Test public void testNegativeSubscription() { BufferedEmittingPublisher<Long> publisher = new BufferedEmittingPublisher<Long>(){}; TestSubscriber<Long> subscriber = new TestSubscriber<Long>() { @Override public void onSubscribe(Subscription subscription) { subscription.request(-1); } }; publisher.subscribe(subscriber); assertThat(subscriber.isComplete(), is(equalTo(false))); assertThat(subscriber.getLastError(), is(not(nullValue()))); assertThat(subscriber.getLastError(), is(instanceOf(IllegalArgumentException.class))); }
@Override public void subscribe(final Flow.Subscriber<? super T> subscriber) { emitter.onSubscribe(() -> state.get().drain(this)); emitter.onRequest((n, cnt) -> { if (requestCallback != null) { requestCallback.accept(n, cnt); } state.get().drain(this); }); emitter.onCancel(() -> state.compareAndSet(State.READY_TO_EMIT, State.CANCELLED)); emitter.subscribe(subscriber); }
BufferedEmittingPublisher implements Flow.Publisher<T> { @Override public void subscribe(final Flow.Subscriber<? super T> subscriber) { emitter.onSubscribe(() -> state.get().drain(this)); emitter.onRequest((n, cnt) -> { if (requestCallback != null) { requestCallback.accept(n, cnt); } state.get().drain(this); }); emitter.onCancel(() -> state.compareAndSet(State.READY_TO_EMIT, State.CANCELLED)); emitter.subscribe(subscriber); } }
BufferedEmittingPublisher implements Flow.Publisher<T> { @Override public void subscribe(final Flow.Subscriber<? super T> subscriber) { emitter.onSubscribe(() -> state.get().drain(this)); emitter.onRequest((n, cnt) -> { if (requestCallback != null) { requestCallback.accept(n, cnt); } state.get().drain(this); }); emitter.onCancel(() -> state.compareAndSet(State.READY_TO_EMIT, State.CANCELLED)); emitter.subscribe(subscriber); } protected  BufferedEmittingPublisher(); }
BufferedEmittingPublisher implements Flow.Publisher<T> { @Override public void subscribe(final Flow.Subscriber<? super T> subscriber) { emitter.onSubscribe(() -> state.get().drain(this)); emitter.onRequest((n, cnt) -> { if (requestCallback != null) { requestCallback.accept(n, cnt); } state.get().drain(this); }); emitter.onCancel(() -> state.compareAndSet(State.READY_TO_EMIT, State.CANCELLED)); emitter.subscribe(subscriber); } protected  BufferedEmittingPublisher(); static BufferedEmittingPublisher<T> create(); @Override void subscribe(final Flow.Subscriber<? super T> subscriber); void onRequest(BiConsumer<Long, Long> requestCallback); void onEmit(Consumer<T> onEmitCallback); int emit(final T item); void fail(Throwable throwable); void complete(); void completeNow(); void clearBuffer(Consumer<T> consumer); boolean isUnbounded(); boolean hasRequests(); boolean isCompleted(); boolean isCancelled(); int bufferSize(); }
BufferedEmittingPublisher implements Flow.Publisher<T> { @Override public void subscribe(final Flow.Subscriber<? super T> subscriber) { emitter.onSubscribe(() -> state.get().drain(this)); emitter.onRequest((n, cnt) -> { if (requestCallback != null) { requestCallback.accept(n, cnt); } state.get().drain(this); }); emitter.onCancel(() -> state.compareAndSet(State.READY_TO_EMIT, State.CANCELLED)); emitter.subscribe(subscriber); } protected  BufferedEmittingPublisher(); static BufferedEmittingPublisher<T> create(); @Override void subscribe(final Flow.Subscriber<? super T> subscriber); void onRequest(BiConsumer<Long, Long> requestCallback); void onEmit(Consumer<T> onEmitCallback); int emit(final T item); void fail(Throwable throwable); void complete(); void completeNow(); void clearBuffer(Consumer<T> consumer); boolean isUnbounded(); boolean hasRequests(); boolean isCompleted(); boolean isCancelled(); int bufferSize();  }