class JettyAsyncResponseListener extends Response.Listener.Adapter implements AsyncBody {

  private final HttpRequest httpRequest;
  private final CompletableFuture<JettyHttpResponse<AsyncBody>> asyncResponse;
  private final CompletableFuture<Void> asyncBodyDone;
  private LongConsumer demand;
  private boolean initialConsumeCalled;

  JettyAsyncResponseListener(HttpRequest httpRequest) {
    this.httpRequest = httpRequest;
    asyncResponse = new CompletableFuture<>();
    asyncBodyDone = new CompletableFuture<>();
  }

  @Override
  public synchronized void consume() {
    if (!this.initialConsumeCalled) {
      this.initialConsumeCalled = true;
    }
    if (demand != null) {
      demand.accept(1);
    }
  }

  @Override
  public CompletableFuture<Void> done() {
    return asyncBodyDone;
  }

  @Override
  public void cancel() {<FILL_FUNCTION_BODY>}

  @Override
  public void onHeaders(Response response) {
    asyncResponse.complete(new JettyHttpResponse<>(httpRequest, response, this));
  }

  @Override
  public void onComplete(Result result) {
    asyncBodyDone.complete(null);
  }

  public CompletableFuture<HttpResponse<AsyncBody>> listen(Request request) {
    request.send(this);
    return asyncResponse.thenApply(HttpResponse.class::cast);
  }

  @Override
  public void onBeforeContent(Response response, LongConsumer demand) {
    synchronized (this) {
      if (!this.initialConsumeCalled) {
        this.demand = demand;
        return;
      }
    }
    demand.accept(1);
  }

  @Override
  public void onContent(Response response, ByteBuffer content, Callback callback) {
    try {
      if (!asyncBodyDone.isCancelled()) {
        onContent(content);
        callback.succeeded();
      }
    } catch (Exception e) {
      callback.failed(e);
    }
  }

  /**
   * Implement to consume the content of the chunked response.
   * <p>
   * Each chunk will be passed <b>in order</b> to this function (<code>onContent{callback.succeeded}</code>)
   *
   * @param content the ByteBuffer containing a chunk of the response.
   * @throws Exception in case the downstream consumer throws an exception.
   */
  protected abstract void onContent(ByteBuffer content) throws Exception;

}

    if (!asyncBodyDone.isDone()) {
      asyncBodyDone.cancel(false);
      asyncResponse.thenAccept(r -> r.getResponse().abort(new CancellationException()));
    }
  
<no_super_class>