@Test public void testItemReceptionWithBackPressure() { AssertSubscriber<Integer> test = AssertSubscriber.create(0); StrictMultiSubscriber<Integer> strict = new StrictMultiSubscriber<>(test); Multi.createFrom().range(0, 5) .subscribe().withSubscriber(strict); test.assertNotTerminated().assertHasNotReceivedAnyItem(); test.request(2); test.assertReceived(0, 1).assertNotTerminated(); test.request(3); test.assertReceived(0, 1, 2, 3, 4).assertCompletedSuccessfully(); test.request(2); test.assertReceived(0, 1, 2, 3, 4).assertCompletedSuccessfully(); }
@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();  }