protected void doHttpRequest(final Request request, final ClientResponseCallback callback) {
        // Highly memory inefficient,
        // but buffer the request content to allow it to be replayed for
        // authentication retries
        final Request.Content content = request.getBody();
        if (content instanceof InputStreamRequestContent) {
            InputStreamRequestContent inputStreamRequestContent = (InputStreamRequestContent) content;
            final List<ByteBuffer> buffers = new ArrayList<>();
            while (true) {
                Content.Chunk chunk = inputStreamRequestContent.read();
                if (chunk.isLast()) {
                    break;
                } else {
                    buffers.add(chunk.getByteBuffer());
                }
            }
            request.body(new ByteBufferRequestContent(buffers.toArray(new ByteBuffer[0])));
            buffers.clear();
        }

        inflightRequests.register();
        // execute the request
        request.send(new BufferingResponseListener(httpClient.getMaxContentLength()) {
            @Override
            public void onComplete(Result result) {
                try {
                    Response response = result.getResponse();

                    final Map<String, String> headers = determineHeadersFrom(response);
                    if (result.isFailed()) {

                        // Failure!!!
                        // including Salesforce errors reported as exception
                        // from SalesforceSecurityHandler
                        Throwable failure = result.getFailure();
                        if (failure instanceof SalesforceException) {
                            httpClient.getWorkerPool()
                                    .execute(() -> callback.onResponse(null, headers, (SalesforceException) failure));
                        } else {
                            final String msg = String.format("Unexpected error {%s:%s} executing {%s:%s}", response.getStatus(),
                                    response.getReason(), request.getMethod(),
                                    request.getURI());
                            httpClient.getWorkerPool().execute(() -> callback.onResponse(null, headers,
                                    new SalesforceException(msg, response.getStatus(), failure)));
                        }
                    } else {

                        // HTTP error status
                        final int status = response.getStatus();
                        HttpRequest request
                                = (HttpRequest) ((HttpRequest) result.getRequest()).getConversation()
                                        .getAttribute(SalesforceSecurityHandler.AUTHENTICATION_REQUEST_ATTRIBUTE);

                        if (status == HttpStatus.BAD_REQUEST_400 && request != null) {
                            // parse login error
                            ContentResponse contentResponse
                                    = new HttpContentResponse(response, getContent(), getMediaType(), getEncoding());
                            try {

                                session.parseLoginResponse(contentResponse, getContentAsString());
                                final String msg = String.format("Unexpected Error {%s:%s} executing {%s:%s}", status,
                                        response.getReason(), request.getMethod(), request.getURI());
                                httpClient.getWorkerPool()
                                        .execute(() -> callback.onResponse(null, headers, new SalesforceException(msg, null)));
                            } catch (SalesforceException e) {

                                final String msg = String.format("Error {%s:%s} executing {%s:%s}", status,
                                        response.getReason(), request.getMethod(), request.getURI());
                                httpClient.getWorkerPool().execute(() -> callback.onResponse(null, headers,
                                        new SalesforceException(msg, response.getStatus(), e)));
                            }
                        } else if (status < HttpStatus.OK_200 || status >= HttpStatus.MULTIPLE_CHOICES_300) {
                            // Salesforce HTTP failure!
                            final SalesforceException exception = createRestException(response, getContentAsInputStream());

                            // for APIs that return body on status 400, such as
                            // Composite API we need content as well
                            httpClient.getWorkerPool()
                                    .execute(() -> callback.onResponse(getContentAsInputStream(), headers, exception));
                        } else {

                            // Success!!!
                            httpClient.getWorkerPool()
                                    .execute(() -> callback.onResponse(getContentAsInputStream(), headers, null));
                        }
                    }
                } finally {
                    inflightRequests.arriveAndDeregister();
                }
            }

            @Override
            public InputStream getContentAsInputStream() {
                if (getContent().length == 0) {
                    return null;
                }
                return super.getContentAsInputStream();
            }
        });
    }
@Test
    public void shouldTimeoutWhenRequestsAreStillOngoing() throws Exception {
        client.doHttpRequest(mock(Request.class), (response, headers, exception) -> {
        });

        // the request never completes

        StopWatch watch = new StopWatch();
        // will wait for 1 second
        client.stop();

        final long elapsed = watch.taken();
        assertTrue(elapsed > 900 && elapsed < 1100);
    }