public static CompletableFuture<ByteSource>
getCompleteBody(HttpServerExchange exchange) {
  log.debug("In getCompleteBody.");
  CompletableFuture<ByteSource> finalResult = new CompletableFuture<>();
  try {
    log.debug("Calling receiveFullBytes.");
    exchange.getRequestReceiver().receiveFullBytes(
        // Success case.
        (HttpServerExchange excng, byte[] bytes) -> {
          log.debug("getFullBytes completed with success.");
          finalResult.complete(ByteSource.wrap(bytes));
        },

        // Error case
        (HttpServerExchange exchng, IOException t) -> {
          log.warn("getFullBytes completed with an error:", t);
          finalResult.completeExceptionally(t);
        }
    );
  } catch (Throwable t) {
    log.warn("receiveFullBytes threw an exception:", t);
    finalResult.completeExceptionally(t);
  }

  return finalResult;
}

@Test
public void getCompleteBodyCompletesWithExceptionOnBodyFailure()
throws Exception {
  AtomicBoolean futureRedeemedWithError = new AtomicBoolean(false);
  CountDownLatch waitForInputToFail = new CountDownLatch(1);

  HttpHandler handler = new HttpHandler() {
    @Override
    public void handleRequest(HttpServerExchange exchange) throws Exception {
      log.info("Handler called.");
      exchange.dispatch();
      log.info("Calling getCompleteBody");
      UndertowUtils.getCompleteBody(exchange).exceptionally((Throwable t) -> {
        log.debug("As expected, received an exception:", t);
        futureRedeemedWithError.set(true);
        waitForInputToFail.countDown();
        return null;
      });
    }
  };
  UndertowTestUtils.RunningServer server =
UndertowTestUtils.startUndertowOnFreePort(handler);

  // A Body() implementation that will return a few chunks of data but the fail
  Body failingRequestBody = new Body() {
    private AtomicInteger readCalls = new AtomicInteger(0);
    @Override
    public long getContentLength() {
      return 1000;
    }

    @Override
    public long read(ByteBuffer buffer) throws IOException {
      if (readCalls.getAndIncrement() < 2) {
        byte[] toSend = "some data".getBytes(Charsets.UTF_8);
        buffer.put(toSend);
        return toSend.length;
      } else {
        throw new RuntimeException("Fake error on sending data.");
      }
    }

    @Override
    public void close() throws IOException {
    }
  };

  // A body generator using the above, busted Body.
  BodyGenerator failingBodyGenerator = new BodyGenerator() {
    @Override
    public Body createBody() throws IOException {
      return failingRequestBody;
    }
  };

  // This will start sending data so our handler gets invoked but it
will then fail partway through.
  SimpleAsyncHttpClient client = new SimpleAsyncHttpClient.Builder()
      .setUrl(String.format("http://localhost:%s", server.getPort()))
      .build();

  client.post(failingBodyGenerator);


  // Test hangs here forever
  waitForInputToFail.await();

  assertThat(futureRedeemedWithError.get()).isTrue();
}