@Test public void testOldSubscriptionIsNotCancelled() { s.switchTo(s2); verifyNoMoreInteractions(s1); }
void switchTo(Subscription next) { requireNonNull(next); long effectiveSourceRequested = sourceEmitted; for (;;) { final long currSourceRequested = sourceRequested; assert currSourceRequested != SWITCHING; if (currSourceRequested == CANCELLED) { next.cancel(); break; } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, SWITCHING)) { assert currSourceRequested >= 0 || currSourceRequested == REQUESTED; final long currRequested = requested; if (currRequested < 0) { sourceRequested = CANCELLED; next.request(currRequested); break; } final long delta = currRequested - effectiveSourceRequested; assert delta >= 0; if (delta != 0) { effectiveSourceRequested = currRequested; next.request(delta); } subscription = next; if (sourceRequestedUpdater.compareAndSet(this, SWITCHING, currRequested)) { break; } } } }
SequentialSubscription implements Subscription, Cancellable { void switchTo(Subscription next) { requireNonNull(next); long effectiveSourceRequested = sourceEmitted; for (;;) { final long currSourceRequested = sourceRequested; assert currSourceRequested != SWITCHING; if (currSourceRequested == CANCELLED) { next.cancel(); break; } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, SWITCHING)) { assert currSourceRequested >= 0 || currSourceRequested == REQUESTED; final long currRequested = requested; if (currRequested < 0) { sourceRequested = CANCELLED; next.request(currRequested); break; } final long delta = currRequested - effectiveSourceRequested; assert delta >= 0; if (delta != 0) { effectiveSourceRequested = currRequested; next.request(delta); } subscription = next; if (sourceRequestedUpdater.compareAndSet(this, SWITCHING, currRequested)) { break; } } } } }
SequentialSubscription implements Subscription, Cancellable { void switchTo(Subscription next) { requireNonNull(next); long effectiveSourceRequested = sourceEmitted; for (;;) { final long currSourceRequested = sourceRequested; assert currSourceRequested != SWITCHING; if (currSourceRequested == CANCELLED) { next.cancel(); break; } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, SWITCHING)) { assert currSourceRequested >= 0 || currSourceRequested == REQUESTED; final long currRequested = requested; if (currRequested < 0) { sourceRequested = CANCELLED; next.request(currRequested); break; } final long delta = currRequested - effectiveSourceRequested; assert delta >= 0; if (delta != 0) { effectiveSourceRequested = currRequested; next.request(delta); } subscription = next; if (sourceRequestedUpdater.compareAndSet(this, SWITCHING, currRequested)) { break; } } } } SequentialSubscription();   SequentialSubscription(Subscription delegate); }
SequentialSubscription implements Subscription, Cancellable { void switchTo(Subscription next) { requireNonNull(next); long effectiveSourceRequested = sourceEmitted; for (;;) { final long currSourceRequested = sourceRequested; assert currSourceRequested != SWITCHING; if (currSourceRequested == CANCELLED) { next.cancel(); break; } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, SWITCHING)) { assert currSourceRequested >= 0 || currSourceRequested == REQUESTED; final long currRequested = requested; if (currRequested < 0) { sourceRequested = CANCELLED; next.request(currRequested); break; } final long delta = currRequested - effectiveSourceRequested; assert delta >= 0; if (delta != 0) { effectiveSourceRequested = currRequested; next.request(delta); } subscription = next; if (sourceRequestedUpdater.compareAndSet(this, SWITCHING, currRequested)) { break; } } } } SequentialSubscription();   SequentialSubscription(Subscription delegate); @Override void request(long n); @Override void cancel(); }
SequentialSubscription implements Subscription, Cancellable { void switchTo(Subscription next) { requireNonNull(next); long effectiveSourceRequested = sourceEmitted; for (;;) { final long currSourceRequested = sourceRequested; assert currSourceRequested != SWITCHING; if (currSourceRequested == CANCELLED) { next.cancel(); break; } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, SWITCHING)) { assert currSourceRequested >= 0 || currSourceRequested == REQUESTED; final long currRequested = requested; if (currRequested < 0) { sourceRequested = CANCELLED; next.request(currRequested); break; } final long delta = currRequested - effectiveSourceRequested; assert delta >= 0; if (delta != 0) { effectiveSourceRequested = currRequested; next.request(delta); } subscription = next; if (sourceRequestedUpdater.compareAndSet(this, SWITCHING, currRequested)) { break; } } } } SequentialSubscription();   SequentialSubscription(Subscription delegate); @Override void request(long n); @Override void cancel();  }