in httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/InternalAbstractHttpAsyncClient.java [165:340]
protected <T> Future<T> doExecute(
final HttpHost httpHost,
final AsyncRequestProducer requestProducer,
final AsyncResponseConsumer<T> responseConsumer,
final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
final HttpContext context,
final FutureCallback<T> callback) {
final ComplexFuture<T> future = new ComplexFuture<>(callback);
try {
if (!isRunning()) {
throw new CancellationException("Request execution cancelled");
}
final HttpClientContext clientContext = context != null ? HttpClientContext.adapt(context) : HttpClientContext.create();
requestProducer.sendRequest((request, entityDetails, c) -> {
RequestConfig requestConfig = null;
if (request instanceof Configurable) {
requestConfig = ((Configurable) request).getConfig();
}
if (requestConfig != null) {
clientContext.setRequestConfig(requestConfig);
}
setupContext(clientContext);
final HttpRoute route = determineRoute(
httpHost != null ? httpHost : RoutingSupport.determineHost(request),
clientContext);
final String exchangeId = ExecSupport.getNextExchangeId();
clientContext.setExchangeId(exchangeId);
if (LOG.isDebugEnabled()) {
LOG.debug("{} preparing request execution", exchangeId);
}
final AsyncExecRuntime execRuntime = createAsyncExecRuntime(pushHandlerFactory);
final AsyncExecChain.Scheduler scheduler = this::executeScheduled;
final AsyncExecChain.Scope scope = new AsyncExecChain.Scope(exchangeId, route, request, future,
clientContext, execRuntime, scheduler, new AtomicInteger(1));
final AtomicBoolean outputTerminated = new AtomicBoolean(false);
executeImmediate(
BasicRequestBuilder.copy(request).build(),
entityDetails != null ? new AsyncEntityProducer() {
@Override
public void releaseResources() {
requestProducer.releaseResources();
}
@Override
public void failed(final Exception cause) {
requestProducer.failed(cause);
}
@Override
public boolean isRepeatable() {
return requestProducer.isRepeatable();
}
@Override
public long getContentLength() {
return entityDetails.getContentLength();
}
@Override
public String getContentType() {
return entityDetails.getContentType();
}
@Override
public String getContentEncoding() {
return entityDetails.getContentEncoding();
}
@Override
public boolean isChunked() {
return entityDetails.isChunked();
}
@Override
public Set<String> getTrailerNames() {
return entityDetails.getTrailerNames();
}
@Override
public int available() {
return requestProducer.available();
}
@Override
public void produce(final DataStreamChannel channel) throws IOException {
if (outputTerminated.get()) {
channel.endStream();
return;
}
requestProducer.produce(channel);
}
} : null,
scope,
new AsyncExecCallback() {
@Override
public AsyncDataConsumer handleResponse(
final HttpResponse response,
final EntityDetails entityDetails) throws HttpException, IOException {
if (response.getCode() >= HttpStatus.SC_CLIENT_ERROR) {
outputTerminated.set(true);
requestProducer.releaseResources();
}
responseConsumer.consumeResponse(response, entityDetails, c,
new FutureCallback<T>() {
@Override
public void completed(final T result) {
future.completed(result);
}
@Override
public void failed(final Exception ex) {
future.failed(ex);
}
@Override
public void cancelled() {
future.cancel();
}
});
return entityDetails != null ? responseConsumer : null;
}
@Override
public void handleInformationResponse(
final HttpResponse response) throws HttpException, IOException {
responseConsumer.informationResponse(response, c);
}
@Override
public void completed() {
if (LOG.isDebugEnabled()) {
LOG.debug("{} message exchange successfully completed", exchangeId);
}
try {
execRuntime.releaseEndpoint();
} finally {
responseConsumer.releaseResources();
requestProducer.releaseResources();
}
}
@Override
public void failed(final Exception cause) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} request failed: {}", exchangeId, cause.getMessage());
}
try {
execRuntime.discardEndpoint();
responseConsumer.failed(cause);
} finally {
try {
future.failed(cause);
} finally {
responseConsumer.releaseResources();
requestProducer.releaseResources();
}
}
}
});
}, context);
} catch (final HttpException | IOException | IllegalStateException ex) {
future.failed(ex);
}
return future;
}