@Test public void testFailureReceptionWhenRequestingNegative() { AtomicReference<Integer> item = new AtomicReference<>(); AtomicReference<Throwable> failure = new AtomicReference<>(); @SuppressWarnings("ReactiveStreamsSubscriberImplementation") Subscriber<Integer> subscriber = new Subscriber<Integer>() { @Override public void onSubscribe(Subscription subscription) { subscription.request(-1); } @Override public void onNext(Integer integer) { item.set(integer); } @Override public void onError(Throwable throwable) { failure.set(throwable); } @Override public void onComplete() { } }; StrictMultiSubscriber<Integer> strict = new StrictMultiSubscriber<>(subscriber); Multi.createFrom().items(23).subscribe().withSubscriber(strict); assertThat(item).doesNotHaveValue(23); assertThat(failure.get()).satisfies(t -> assertThat(t).isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("3.9")); }
@Override public void request(long n) { if (n <= 0) { cancel(); onError(new IllegalArgumentException("Reactive Streams Rule 3.9 violated: request must be positive, but was " + n)); } else { Subscriptions.requestIfNotNullOrAccumulate(upstream, requested, n); } }
StrictMultiSubscriber implements MultiSubscriber<T>, Subscription { @Override public void request(long n) { if (n <= 0) { cancel(); onError(new IllegalArgumentException("Reactive Streams Rule 3.9 violated: request must be positive, but was " + n)); } else { Subscriptions.requestIfNotNullOrAccumulate(upstream, requested, n); } } }
StrictMultiSubscriber implements MultiSubscriber<T>, Subscription { @Override public void request(long n) { if (n <= 0) { cancel(); onError(new IllegalArgumentException("Reactive Streams Rule 3.9 violated: request must be positive, but was " + n)); } else { Subscriptions.requestIfNotNullOrAccumulate(upstream, requested, n); } } StrictMultiSubscriber(Subscriber<? super T> downstream); }
StrictMultiSubscriber implements MultiSubscriber<T>, Subscription { @Override public void request(long n) { if (n <= 0) { cancel(); onError(new IllegalArgumentException("Reactive Streams Rule 3.9 violated: request must be positive, but was " + n)); } else { Subscriptions.requestIfNotNullOrAccumulate(upstream, requested, n); } } StrictMultiSubscriber(Subscriber<? super T> downstream); @Override void request(long n); @Override void cancel(); @Override void onSubscribe(Subscription s); @Override void onItem(T t); @Override void onFailure(Throwable t); @Override void onCompletion(); }
StrictMultiSubscriber implements MultiSubscriber<T>, Subscription { @Override public void request(long n) { if (n <= 0) { cancel(); onError(new IllegalArgumentException("Reactive Streams Rule 3.9 violated: request must be positive, but was " + n)); } else { Subscriptions.requestIfNotNullOrAccumulate(upstream, requested, n); } } StrictMultiSubscriber(Subscriber<? super T> downstream); @Override void request(long n); @Override void cancel(); @Override void onSubscribe(Subscription s); @Override void onItem(T t); @Override void onFailure(Throwable t); @Override void onCompletion();  }