class VertxHttpResponse extends StandardHttpHeaders implements HttpResponse<AsyncBody> {
    private final AsyncBody result;
    private final HttpClientResponse resp;
    private final HttpRequest request;

    private VertxHttpResponse(AsyncBody result, HttpClientResponse resp, HttpRequest request) {
      super(toHeadersMap(resp.headers()));
      this.result = result;
      this.resp = resp;
      this.request = request;
    }

    @Override
    public int code() {
      return resp.statusCode();
    }

    @Override
    public AsyncBody body() {
      return result;
    }

    @Override
    public HttpRequest request() {
      return request;
    }

    @Override
    public Optional<HttpResponse<?>> previousResponse() {
      return Optional.empty();
    }
  }

  final Vertx vertx;
  private final RequestOptions options;
  private StandardHttpRequest request;

  public VertxHttpRequest(Vertx vertx, RequestOptions options, StandardHttpRequest request) {
    this.vertx = vertx;
    this.options = options;
    this.request = request;
  }

  public CompletableFuture<HttpResponse<AsyncBody>> consumeBytes(HttpClient client,
      AsyncBody.Consumer<List<ByteBuffer>> consumer) {
    Function<HttpClientResponse, HttpResponse<AsyncBody>> responseHandler = resp -> {
      resp.pause();
      AsyncBody result = new AsyncBody() {
        final CompletableFuture<Void> done = new CompletableFuture<>();

        @Override
        public void consume() {
          resp.fetch(1);
        }

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

        @Override
        public void cancel() {
          resp.handler(null);
          resp.endHandler(null);
          resp.request().reset();
          done.cancel(false);
        }

      };
      resp.handler(buffer -> {
        try {
          consumer.consume(Arrays.asList(ByteBuffer.wrap(buffer.getBytes())), result);
        } catch (Exception e) {
          resp.request().reset();
          result.done().completeExceptionally(e);
        }
      }).endHandler(end -> result.done().complete(null));
      return new VertxHttpResponse(result, resp, request);
    };
    return client.request(options).compose(request -> {
      Future<HttpClientResponse> fut;
      BodyContent body = this.request.body();
      if (body != null) {
        if (body instanceof StandardHttpRequest.StringBodyContent) {
          Buffer buffer = Buffer.buffer(((StandardHttpRequest.StringBodyContent) body).getContent());
          fut = request.send(buffer);
        } else if (body instanceof StandardHttpRequest.ByteArrayBodyContent) {
          Buffer buffer = Buffer.buffer(((StandardHttpRequest.ByteArrayBodyContent) body).getContent());
          fut = request.send(buffer);
        } else if (body instanceof StandardHttpRequest.InputStreamBodyContent) {
          StandardHttpRequest.InputStreamBodyContent bodyContent = (StandardHttpRequest.InputStreamBodyContent) body;
          InputStream is = bodyContent.getContent();

          ReadStream<Buffer> stream = new InputStreamReadStream(this, is, request);
          fut = request.send(stream);
        } else {
          fut = Future.failedFuture("Unsupported body content");
        }
      } else {
        fut = request.send();
      }
      return fut.map(responseHandler);
    }).toCompletionStage().toCompletableFuture();
  }

  static Map<String, List<String>> toHeadersMap(MultiMap multiMap) {<FILL_FUNCTION_BODY>

    Map<String, List<String>> headers = new LinkedHashMap<>();
    multiMap.names().forEach(k -> headers.put(k, multiMap.getAll(k)));
    return headers;
  
<no_super_class>