@Test public void testPipedStreams() throws Exception { try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) { PipedOutputStream pout = new PipedOutputStream(); try (PipedInputStream pin = new PipedInputStream(pout)) { BodyDeferringAsyncHandler handler = new BodyDeferringAsyncHandler(pout); ListenableFuture<Response> respFut = client.prepareGet(getTargetUrl()).execute(handler); Response resp = handler.getResponse(); if (resp.getStatusCode() == 200) { try (BodyDeferringInputStream is = new BodyDeferringInputStream(respFut, handler, pin)) { String body = IOUtils.toString(is, StandardCharsets.UTF_8); assertTrue(body.contains("ABCDEF")); } } else { throw new IOException("HTTP error " + resp.getStatusCode()); } } } }
public Response getResponse() throws InterruptedException, IOException { headersArrived.await(); try { semaphore.acquire(); if (throwable != null) { throw new IOException(throwable.getMessage(), throwable); } else { return response; } } finally { semaphore.release(); } }
BodyDeferringAsyncHandler implements AsyncHandler<Response> { public Response getResponse() throws InterruptedException, IOException { headersArrived.await(); try { semaphore.acquire(); if (throwable != null) { throw new IOException(throwable.getMessage(), throwable); } else { return response; } } finally { semaphore.release(); } } }
BodyDeferringAsyncHandler implements AsyncHandler<Response> { public Response getResponse() throws InterruptedException, IOException { headersArrived.await(); try { semaphore.acquire(); if (throwable != null) { throw new IOException(throwable.getMessage(), throwable); } else { return response; } } finally { semaphore.release(); } } BodyDeferringAsyncHandler(final OutputStream os); }
BodyDeferringAsyncHandler implements AsyncHandler<Response> { public Response getResponse() throws InterruptedException, IOException { headersArrived.await(); try { semaphore.acquire(); if (throwable != null) { throw new IOException(throwable.getMessage(), throwable); } else { return response; } } finally { semaphore.release(); } } BodyDeferringAsyncHandler(final OutputStream os); @Override void onThrowable(Throwable t); @Override State onStatusReceived(HttpResponseStatus responseStatus); @Override State onHeadersReceived(HttpHeaders headers); @Override State onTrailingHeadersReceived(HttpHeaders headers); @Override void onRetry(); @Override State onBodyPartReceived(HttpResponseBodyPart bodyPart); @Override Response onCompleted(); Response getResponse(); }
BodyDeferringAsyncHandler implements AsyncHandler<Response> { public Response getResponse() throws InterruptedException, IOException { headersArrived.await(); try { semaphore.acquire(); if (throwable != null) { throw new IOException(throwable.getMessage(), throwable); } else { return response; } } finally { semaphore.release(); } } BodyDeferringAsyncHandler(final OutputStream os); @Override void onThrowable(Throwable t); @Override State onStatusReceived(HttpResponseStatus responseStatus); @Override State onHeadersReceived(HttpHeaders headers); @Override State onTrailingHeadersReceived(HttpHeaders headers); @Override void onRetry(); @Override State onBodyPartReceived(HttpResponseBodyPart bodyPart); @Override Response onCompleted(); Response getResponse();  }