code: private void deliverUnsafe() {
      // Try to acquire the lock
      if (lock.getAndIncrement() != 0) {
        return;
      }
  
      do {
        // Optimization: the inner loop will eager process any accumulated state, so reset the lock
        // for just this iteration. (If another event occurs during processing, it can increment the
        // lock to enqueue another iteration).
        lock.lazySet(1);
  
        // Process the upstream message if one exists.
        pollUpstream();
  
        // Eagerly notify of onComplete/onError disregarding demand.
        // NOTE: this will purposely leave wip set to avoid further deliveries.
        if (maybeFinish()) {
          return;
        }
  
        // Deliver as many messages as possible
        int demandSnapshot = numRequested.get();
        int delivered = 0;
  
        while (delivered < demandSnapshot) {
          // Deliver a message if we can
          if (reframer.hasFullFrame()) {
            delivered++;
            outerResponseObserver.onResponse(reframer.pop());
          } else {
            // Otherwise request more from upstream (if we haven't done so already)
            if (!awaitingInner) {
              awaitingInner = true;
              innerController.request(1);
            }
            break;
          }
  
          if (maybeFinish()) {
            return;
          }
        }
  
        // update the counter in bulk
        if (delivered != 0) {
          numRequested.addAndGet(-delivered);
        }
      } while (lock.decrementAndGet() != 0);
    }
docs: Coordinates state transfer between inner/downstream callable and the outer/upstream. It
  orchestrates the flow of demand from downstream to upstream. The data flows from upstream
  through the Reframer to downstream. It is back pressure aware and will only send as many
  messages as were requested. However, it will send unsolicited onComplete & onError messages.
  
  <p>This method is thread safe and performs all state changes (including interactions with the
  Reframer) using CAS mutex.