### Question:
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);  }### Answer:
@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); }
@Test public void testDrainLoopWithMissedRequestsAndItems() { switchable = new SwitchableSubscriptionSubscriber<Integer>(downstream) { @Override public void onItem(Integer item) { downstream.onItem(item); } }; switchable.wip.getAndIncrement(); switchable.requested = 0; switchable.missedRequested.set(Long.MAX_VALUE); switchable.missedItems.set(25); switchable.drainLoop(); assertThat(switchable.requested).isEqualTo(Long.MAX_VALUE); }
@Test public void testWhenWeEmitMoreThanRequested() { 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.missedItems.set(2); switchable.drainLoop(); assertThat(switchable.requested).isEqualTo(0); assertThat(switchable.missedRequested).hasValue(0); assertThat(switchable.missedItems).hasValue(0); }
@Test public void testMissedFirstSubscription() { switchable = new SwitchableSubscriptionSubscriber<Integer>(downstream) { @Override public void onItem(Integer item) { downstream.onItem(item); } }; switchable.wip.getAndIncrement(); Subscription subscription = mock(Subscription.class); switchable.pendingSubscription.set(subscription); switchable.drainLoop(); assertThat(subscription).isEqualTo(switchable.currentUpstream.get()); }