@Test public void testMissedRequests() { switchable = new SwitchableSubscriptionSubscriber<Integer>(downstream) { @Override public void onItem(Integer item) { downstream.onItem(item); } }; switchable.wip.getAndIncrement(); switchable.requested = 0; switchable.missedRequested.set(1); switchable.drainLoop(); assertThat(switchable.requested).isEqualTo(1); }
void drainLoop() { int missed = 1; long requestAmount = 0L; Subscription requestTarget = null; for (;;) { Subscription nextUpstream = pendingSubscription.getAndSet(null); long pendingRequests = missedRequested.getAndSet(0L); long pendingItems = missedItems.getAndSet(0L); Subscription upstream = currentUpstream.get(); if (cancelled.get()) { if (upstream != null) { upstream.cancel(); currentUpstream.set(null); } if (nextUpstream != null) { nextUpstream.cancel(); } } else { long req = requested; if (req != Long.MAX_VALUE) { long res = Subscriptions.add(req, pendingRequests); if (res != Long.MAX_VALUE) { long remaining = res - pendingItems; if (remaining < 0L) { remaining = 0; } req = remaining; } else { req = res; } requested = req; } if (nextUpstream != null) { if (upstream != null && cancelUpstreamOnSwitch()) { upstream.cancel(); } currentUpstream.set(nextUpstream); if (req != 0L) { requestAmount = Subscriptions.add(requestAmount, req); requestTarget = nextUpstream; } } else if (pendingRequests != 0L && upstream != null) { requestAmount = Subscriptions.add(requestAmount, pendingRequests); requestTarget = upstream; } } missed = wip.addAndGet(-missed); if (missed == 0) { if (requestAmount != 0L) { requestTarget.request(requestAmount); } return; } } }
SwitchableSubscriptionSubscriber implements MultiSubscriber<O>, Subscription { void drainLoop() { int missed = 1; long requestAmount = 0L; Subscription requestTarget = null; for (;;) { Subscription nextUpstream = pendingSubscription.getAndSet(null); long pendingRequests = missedRequested.getAndSet(0L); long pendingItems = missedItems.getAndSet(0L); Subscription upstream = currentUpstream.get(); if (cancelled.get()) { if (upstream != null) { upstream.cancel(); currentUpstream.set(null); } if (nextUpstream != null) { nextUpstream.cancel(); } } else { long req = requested; if (req != Long.MAX_VALUE) { long res = Subscriptions.add(req, pendingRequests); if (res != Long.MAX_VALUE) { long remaining = res - pendingItems; if (remaining < 0L) { remaining = 0; } req = remaining; } else { req = res; } requested = req; } if (nextUpstream != null) { if (upstream != null && cancelUpstreamOnSwitch()) { upstream.cancel(); } currentUpstream.set(nextUpstream); if (req != 0L) { requestAmount = Subscriptions.add(requestAmount, req); requestTarget = nextUpstream; } } else if (pendingRequests != 0L && upstream != null) { requestAmount = Subscriptions.add(requestAmount, pendingRequests); requestTarget = upstream; } } missed = wip.addAndGet(-missed); if (missed == 0) { if (requestAmount != 0L) { requestTarget.request(requestAmount); } return; } } } }
SwitchableSubscriptionSubscriber implements MultiSubscriber<O>, Subscription { void drainLoop() { int missed = 1; long requestAmount = 0L; Subscription requestTarget = null; for (;;) { Subscription nextUpstream = pendingSubscription.getAndSet(null); long pendingRequests = missedRequested.getAndSet(0L); long pendingItems = missedItems.getAndSet(0L); Subscription upstream = currentUpstream.get(); if (cancelled.get()) { if (upstream != null) { upstream.cancel(); currentUpstream.set(null); } if (nextUpstream != null) { nextUpstream.cancel(); } } else { long req = requested; if (req != Long.MAX_VALUE) { long res = Subscriptions.add(req, pendingRequests); if (res != Long.MAX_VALUE) { long remaining = res - pendingItems; if (remaining < 0L) { remaining = 0; } req = remaining; } else { req = res; } requested = req; } if (nextUpstream != null) { if (upstream != null && cancelUpstreamOnSwitch()) { upstream.cancel(); } currentUpstream.set(nextUpstream); if (req != 0L) { requestAmount = Subscriptions.add(requestAmount, req); requestTarget = nextUpstream; } } else if (pendingRequests != 0L && upstream != null) { requestAmount = Subscriptions.add(requestAmount, pendingRequests); requestTarget = upstream; } } missed = wip.addAndGet(-missed); if (missed == 0) { if (requestAmount != 0L) { requestTarget.request(requestAmount); } return; } } } SwitchableSubscriptionSubscriber(MultiSubscriber<? super O> downstream); }
SwitchableSubscriptionSubscriber implements MultiSubscriber<O>, Subscription { void drainLoop() { int missed = 1; long requestAmount = 0L; Subscription requestTarget = null; for (;;) { Subscription nextUpstream = pendingSubscription.getAndSet(null); long pendingRequests = missedRequested.getAndSet(0L); long pendingItems = missedItems.getAndSet(0L); Subscription upstream = currentUpstream.get(); if (cancelled.get()) { if (upstream != null) { upstream.cancel(); currentUpstream.set(null); } if (nextUpstream != null) { nextUpstream.cancel(); } } else { long req = requested; if (req != Long.MAX_VALUE) { long res = Subscriptions.add(req, pendingRequests); if (res != Long.MAX_VALUE) { long remaining = res - pendingItems; if (remaining < 0L) { remaining = 0; } req = remaining; } else { req = res; } requested = req; } if (nextUpstream != null) { if (upstream != null && cancelUpstreamOnSwitch()) { upstream.cancel(); } currentUpstream.set(nextUpstream); if (req != 0L) { requestAmount = Subscriptions.add(requestAmount, req); requestTarget = nextUpstream; } } else if (pendingRequests != 0L && upstream != null) { requestAmount = Subscriptions.add(requestAmount, pendingRequests); requestTarget = upstream; } } missed = wip.addAndGet(-missed); if (missed == 0) { if (requestAmount != 0L) { requestTarget.request(requestAmount); } return; } } } SwitchableSubscriptionSubscriber(MultiSubscriber<? super O> downstream); @Override void cancel(); boolean isCancelled(); @Override void onCompletion(); @Override void onFailure(Throwable t); @Override void onSubscribe(Subscription s); void emitted(long n); @Override final void request(long n); }
SwitchableSubscriptionSubscriber implements MultiSubscriber<O>, Subscription { void drainLoop() { int missed = 1; long requestAmount = 0L; Subscription requestTarget = null; for (;;) { Subscription nextUpstream = pendingSubscription.getAndSet(null); long pendingRequests = missedRequested.getAndSet(0L); long pendingItems = missedItems.getAndSet(0L); Subscription upstream = currentUpstream.get(); if (cancelled.get()) { if (upstream != null) { upstream.cancel(); currentUpstream.set(null); } if (nextUpstream != null) { nextUpstream.cancel(); } } else { long req = requested; if (req != Long.MAX_VALUE) { long res = Subscriptions.add(req, pendingRequests); if (res != Long.MAX_VALUE) { long remaining = res - pendingItems; if (remaining < 0L) { remaining = 0; } req = remaining; } else { req = res; } requested = req; } if (nextUpstream != null) { if (upstream != null && cancelUpstreamOnSwitch()) { upstream.cancel(); } currentUpstream.set(nextUpstream); if (req != 0L) { requestAmount = Subscriptions.add(requestAmount, req); requestTarget = nextUpstream; } } else if (pendingRequests != 0L && upstream != null) { requestAmount = Subscriptions.add(requestAmount, pendingRequests); requestTarget = upstream; } } missed = wip.addAndGet(-missed); if (missed == 0) { if (requestAmount != 0L) { requestTarget.request(requestAmount); } return; } } } SwitchableSubscriptionSubscriber(MultiSubscriber<? super O> downstream); @Override void cancel(); boolean isCancelled(); @Override void onCompletion(); @Override void onFailure(Throwable t); @Override void onSubscribe(Subscription s); void emitted(long n); @Override final void request(long n);  }