@Test public void testRequestNIncremental() { s.request(1); verify(s1).request(1); verifyNoMoreInteractions(s1); s.request(3); verify(s1).request(3); verifyNoMoreInteractions(s1); }
@Override public void request(long n) { final long currRequested; if (isRequestNValid(n)) { currRequested = requestedUpdater.accumulateAndGet(this, n, FlowControlUtils::addWithOverflowProtectionIfNotNegative); } else { currRequested = sanitizeInvalidRequestN(n); requested = currRequested; } for (;;) { final long currSourceRequested = sourceRequested; if (currSourceRequested == CANCELLED) { break; } else if (currSourceRequested < 0) { assert currSourceRequested == SWITCHING || currSourceRequested == REQUESTED; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, REQUESTED)) { break; } } else { final Subscription currSubscription = subscription; if (isRequestNValid(currRequested)) { final long delta = currRequested - currSourceRequested; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, currSourceRequested + delta)) { if (delta != 0) { currSubscription.request(delta); } break; } } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, CANCELLED)) { currSubscription.request(currRequested); break; } } } }
SequentialSubscription implements Subscription, Cancellable { @Override public void request(long n) { final long currRequested; if (isRequestNValid(n)) { currRequested = requestedUpdater.accumulateAndGet(this, n, FlowControlUtils::addWithOverflowProtectionIfNotNegative); } else { currRequested = sanitizeInvalidRequestN(n); requested = currRequested; } for (;;) { final long currSourceRequested = sourceRequested; if (currSourceRequested == CANCELLED) { break; } else if (currSourceRequested < 0) { assert currSourceRequested == SWITCHING || currSourceRequested == REQUESTED; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, REQUESTED)) { break; } } else { final Subscription currSubscription = subscription; if (isRequestNValid(currRequested)) { final long delta = currRequested - currSourceRequested; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, currSourceRequested + delta)) { if (delta != 0) { currSubscription.request(delta); } break; } } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, CANCELLED)) { currSubscription.request(currRequested); break; } } } } }
SequentialSubscription implements Subscription, Cancellable { @Override public void request(long n) { final long currRequested; if (isRequestNValid(n)) { currRequested = requestedUpdater.accumulateAndGet(this, n, FlowControlUtils::addWithOverflowProtectionIfNotNegative); } else { currRequested = sanitizeInvalidRequestN(n); requested = currRequested; } for (;;) { final long currSourceRequested = sourceRequested; if (currSourceRequested == CANCELLED) { break; } else if (currSourceRequested < 0) { assert currSourceRequested == SWITCHING || currSourceRequested == REQUESTED; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, REQUESTED)) { break; } } else { final Subscription currSubscription = subscription; if (isRequestNValid(currRequested)) { final long delta = currRequested - currSourceRequested; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, currSourceRequested + delta)) { if (delta != 0) { currSubscription.request(delta); } break; } } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, CANCELLED)) { currSubscription.request(currRequested); break; } } } } SequentialSubscription();   SequentialSubscription(Subscription delegate); }
SequentialSubscription implements Subscription, Cancellable { @Override public void request(long n) { final long currRequested; if (isRequestNValid(n)) { currRequested = requestedUpdater.accumulateAndGet(this, n, FlowControlUtils::addWithOverflowProtectionIfNotNegative); } else { currRequested = sanitizeInvalidRequestN(n); requested = currRequested; } for (;;) { final long currSourceRequested = sourceRequested; if (currSourceRequested == CANCELLED) { break; } else if (currSourceRequested < 0) { assert currSourceRequested == SWITCHING || currSourceRequested == REQUESTED; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, REQUESTED)) { break; } } else { final Subscription currSubscription = subscription; if (isRequestNValid(currRequested)) { final long delta = currRequested - currSourceRequested; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, currSourceRequested + delta)) { if (delta != 0) { currSubscription.request(delta); } break; } } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, CANCELLED)) { currSubscription.request(currRequested); break; } } } } SequentialSubscription();   SequentialSubscription(Subscription delegate); @Override void request(long n); @Override void cancel(); }
SequentialSubscription implements Subscription, Cancellable { @Override public void request(long n) { final long currRequested; if (isRequestNValid(n)) { currRequested = requestedUpdater.accumulateAndGet(this, n, FlowControlUtils::addWithOverflowProtectionIfNotNegative); } else { currRequested = sanitizeInvalidRequestN(n); requested = currRequested; } for (;;) { final long currSourceRequested = sourceRequested; if (currSourceRequested == CANCELLED) { break; } else if (currSourceRequested < 0) { assert currSourceRequested == SWITCHING || currSourceRequested == REQUESTED; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, REQUESTED)) { break; } } else { final Subscription currSubscription = subscription; if (isRequestNValid(currRequested)) { final long delta = currRequested - currSourceRequested; if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, currSourceRequested + delta)) { if (delta != 0) { currSubscription.request(delta); } break; } } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, CANCELLED)) { currSubscription.request(currRequested); break; } } } } SequentialSubscription();   SequentialSubscription(Subscription delegate); @Override void request(long n); @Override void cancel();  }