You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2017/11/13 21:45:58 UTC
[3/6] httpcomponents-client git commit: * HTTP/2 multiplexing
HttpAsyncClient implementation * Restructured integration tests to reduce
test duplication
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/AsyncMainClientExec.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/AsyncMainClientExec.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/AsyncMainClientExec.java
deleted file mode 100644
index d23f20f..0000000
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/AsyncMainClientExec.java
+++ /dev/null
@@ -1,229 +0,0 @@
-/*
- * ====================================================================
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- * ====================================================================
- *
- * This software consists of voluntary contributions made by many
- * individuals on behalf of the Apache Software Foundation. For more
- * information on the Apache Software Foundation, please see
- * <http://www.apache.org/>.
- *
- */
-package org.apache.hc.client5.http.impl.async;
-
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.nio.ByteBuffer;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.hc.client5.http.ConnectionKeepAliveStrategy;
-import org.apache.hc.client5.http.HttpRoute;
-import org.apache.hc.client5.http.UserTokenHandler;
-import org.apache.hc.client5.http.async.AsyncExecCallback;
-import org.apache.hc.client5.http.async.AsyncExecChain;
-import org.apache.hc.client5.http.async.AsyncExecChainHandler;
-import org.apache.hc.client5.http.async.AsyncExecRuntime;
-import org.apache.hc.client5.http.protocol.HttpClientContext;
-import org.apache.hc.core5.http.EntityDetails;
-import org.apache.hc.core5.http.Header;
-import org.apache.hc.core5.http.HttpException;
-import org.apache.hc.core5.http.HttpRequest;
-import org.apache.hc.core5.http.HttpResponse;
-import org.apache.hc.core5.http.HttpStatus;
-import org.apache.hc.core5.http.message.RequestLine;
-import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
-import org.apache.hc.core5.http.nio.AsyncDataConsumer;
-import org.apache.hc.core5.http.nio.AsyncEntityProducer;
-import org.apache.hc.core5.http.nio.CapacityChannel;
-import org.apache.hc.core5.http.nio.DataStreamChannel;
-import org.apache.hc.core5.http.nio.RequestChannel;
-import org.apache.hc.core5.util.TimeValue;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
-class AsyncMainClientExec implements AsyncExecChainHandler {
-
- private final Logger log = LogManager.getLogger(getClass());
-
- private final ConnectionKeepAliveStrategy keepAliveStrategy;
- private final UserTokenHandler userTokenHandler;
-
- AsyncMainClientExec(final ConnectionKeepAliveStrategy keepAliveStrategy, final UserTokenHandler userTokenHandler) {
- this.keepAliveStrategy = keepAliveStrategy;
- this.userTokenHandler = userTokenHandler;
- }
-
- @Override
- public void execute(
- final HttpRequest request,
- final AsyncEntityProducer entityProducer,
- final AsyncExecChain.Scope scope,
- final AsyncExecChain chain,
- final AsyncExecCallback asyncExecCallback) throws HttpException, IOException {
- final String exchangeId = scope.exchangeId;
- final HttpRoute route = scope.route;
- final HttpClientContext clientContext = scope.clientContext;
- final AsyncExecRuntime execRuntime = scope.execRuntime;
-
- if (log.isDebugEnabled()) {
- log.debug(exchangeId + ": executing " + new RequestLine(request));
- }
-
- final AtomicInteger messageCountDown = new AtomicInteger(2);
- final AsyncClientExchangeHandler internalExchangeHandler = new AsyncClientExchangeHandler() {
-
- private final AtomicReference<AsyncDataConsumer> entityConsumerRef = new AtomicReference<>(null);
-
- @Override
- public void releaseResources() {
- final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
- if (entityConsumer != null) {
- entityConsumer.releaseResources();
- }
- }
-
- @Override
- public void failed(final Exception cause) {
- execRuntime.markConnectionNonReusable();
- asyncExecCallback.failed(cause);
- }
-
- @Override
- public void cancel() {
- failed(new InterruptedIOException());
- }
-
- @Override
- public void produceRequest(final RequestChannel channel) throws HttpException, IOException {
- channel.sendRequest(request, entityProducer);
- if (entityProducer == null) {
- messageCountDown.decrementAndGet();
- }
- }
-
- @Override
- public int available() {
- return entityProducer.available();
- }
-
- @Override
- public void produce(final DataStreamChannel channel) throws IOException {
- entityProducer.produce(new DataStreamChannel() {
-
- @Override
- public void requestOutput() {
- channel.requestOutput();
- }
-
- @Override
- public int write(final ByteBuffer src) throws IOException {
- return channel.write(src);
- }
-
- @Override
- public void endStream(final List<? extends Header> trailers) throws IOException {
- channel.endStream(trailers);
- if (messageCountDown.decrementAndGet() <= 0) {
- asyncExecCallback.completed();
- }
- }
-
- @Override
- public void endStream() throws IOException {
- channel.endStream();
- if (messageCountDown.decrementAndGet() <= 0) {
- asyncExecCallback.completed();
- }
- }
-
- });
- }
-
- @Override
- public void consumeInformation(final HttpResponse response) throws HttpException, IOException {
- }
-
- @Override
- public void consumeResponse(final HttpResponse response, final EntityDetails entityDetails) throws HttpException, IOException {
- entityConsumerRef.set(asyncExecCallback.handleResponse(response, entityDetails));
- if (response.getCode() >= HttpStatus.SC_CLIENT_ERROR) {
- messageCountDown.decrementAndGet();
- }
- final TimeValue keepAliveDuration = keepAliveStrategy.getKeepAliveDuration(response, clientContext);
- Object userToken = clientContext.getUserToken();
- if (userToken == null) {
- userToken = userTokenHandler.getUserToken(route, clientContext);
- clientContext.setAttribute(HttpClientContext.USER_TOKEN, userToken);
- }
- execRuntime.markConnectionReusable(userToken, keepAliveDuration);
- if (entityDetails == null) {
- execRuntime.validateConnection();
- if (messageCountDown.decrementAndGet() <= 0) {
- asyncExecCallback.completed();
- }
- }
- }
-
- @Override
- public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
- final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
- if (entityConsumer != null) {
- entityConsumer.updateCapacity(capacityChannel);
- } else {
- capacityChannel.update(Integer.MAX_VALUE);
- }
- }
-
- @Override
- public int consume(final ByteBuffer src) throws IOException {
- final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
- if (entityConsumer != null) {
- return entityConsumer.consume(src);
- } else {
- return Integer.MAX_VALUE;
- }
- }
-
- @Override
- public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
- final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
- if (entityConsumer != null) {
- entityConsumer.streamEnd(trailers);
- } else {
- execRuntime.validateConnection();
- }
- if (messageCountDown.decrementAndGet() <= 0) {
- asyncExecCallback.completed();
- }
- }
-
- };
-
- if (log.isDebugEnabled()) {
- execRuntime.execute(
- new LoggingAsyncClientExchangeHandler(log, exchangeId, internalExchangeHandler),
- clientContext);
- } else {
- execRuntime.execute(
- internalExchangeHandler, clientContext);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientBuilder.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientBuilder.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientBuilder.java
new file mode 100644
index 0000000..1728709
--- /dev/null
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientBuilder.java
@@ -0,0 +1,885 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.client5.http.impl.async;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.hc.client5.http.AuthenticationStrategy;
+import org.apache.hc.client5.http.DnsResolver;
+import org.apache.hc.client5.http.HttpRequestRetryHandler;
+import org.apache.hc.client5.http.SchemePortResolver;
+import org.apache.hc.client5.http.SystemDefaultDnsResolver;
+import org.apache.hc.client5.http.async.AsyncExecChainHandler;
+import org.apache.hc.client5.http.auth.AuthSchemeProvider;
+import org.apache.hc.client5.http.auth.CredentialsProvider;
+import org.apache.hc.client5.http.auth.KerberosConfig;
+import org.apache.hc.client5.http.config.AuthSchemes;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.cookie.BasicCookieStore;
+import org.apache.hc.client5.http.cookie.CookieSpecProvider;
+import org.apache.hc.client5.http.cookie.CookieStore;
+import org.apache.hc.client5.http.impl.ChainElements;
+import org.apache.hc.client5.http.impl.CookieSpecRegistries;
+import org.apache.hc.client5.http.impl.DefaultAuthenticationStrategy;
+import org.apache.hc.client5.http.impl.DefaultHttpRequestRetryHandler;
+import org.apache.hc.client5.http.impl.DefaultRedirectStrategy;
+import org.apache.hc.client5.http.impl.DefaultSchemePortResolver;
+import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
+import org.apache.hc.client5.http.impl.auth.BasicSchemeFactory;
+import org.apache.hc.client5.http.impl.auth.CredSspSchemeFactory;
+import org.apache.hc.client5.http.impl.auth.DigestSchemeFactory;
+import org.apache.hc.client5.http.impl.auth.KerberosSchemeFactory;
+import org.apache.hc.client5.http.impl.auth.NTLMSchemeFactory;
+import org.apache.hc.client5.http.impl.auth.SPNegoSchemeFactory;
+import org.apache.hc.client5.http.impl.auth.SystemDefaultCredentialsProvider;
+import org.apache.hc.client5.http.impl.nio.MultuhomeConnectionInitiator;
+import org.apache.hc.client5.http.impl.routing.DefaultRoutePlanner;
+import org.apache.hc.client5.http.protocol.RedirectStrategy;
+import org.apache.hc.client5.http.protocol.RequestAddCookies;
+import org.apache.hc.client5.http.protocol.RequestAuthCache;
+import org.apache.hc.client5.http.protocol.RequestDefaultHeaders;
+import org.apache.hc.client5.http.protocol.RequestExpectContinue;
+import org.apache.hc.client5.http.protocol.ResponseProcessCookies;
+import org.apache.hc.client5.http.routing.HttpRoutePlanner;
+import org.apache.hc.client5.http.ssl.H2TlsStrategy;
+import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.concurrent.DefaultThreadFactory;
+import org.apache.hc.core5.function.Callback;
+import org.apache.hc.core5.function.Resolver;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.HttpRequestInterceptor;
+import org.apache.hc.core5.http.HttpResponseInterceptor;
+import org.apache.hc.core5.http.config.CharCodingConfig;
+import org.apache.hc.core5.http.config.Lookup;
+import org.apache.hc.core5.http.config.NamedElementChain;
+import org.apache.hc.core5.http.config.RegistryBuilder;
+import org.apache.hc.core5.http.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http.nio.HandlerFactory;
+import org.apache.hc.core5.http.nio.command.ShutdownCommand;
+import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
+import org.apache.hc.core5.http.protocol.DefaultHttpProcessor;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.http.protocol.HttpProcessor;
+import org.apache.hc.core5.http.protocol.HttpProcessorBuilder;
+import org.apache.hc.core5.http.protocol.RequestTargetHost;
+import org.apache.hc.core5.http.protocol.RequestUserAgent;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.nio.pool.H2ConnPool;
+import org.apache.hc.core5.http2.protocol.H2RequestConnControl;
+import org.apache.hc.core5.http2.protocol.H2RequestContent;
+import org.apache.hc.core5.http2.protocol.H2RequestTargetHost;
+import org.apache.hc.core5.io.ShutdownType;
+import org.apache.hc.core5.reactor.DefaultConnectingIOReactor;
+import org.apache.hc.core5.reactor.IOEventHandlerFactory;
+import org.apache.hc.core5.reactor.IOReactorConfig;
+import org.apache.hc.core5.reactor.IOSession;
+import org.apache.hc.core5.util.Args;
+import org.apache.hc.core5.util.TimeValue;
+import org.apache.hc.core5.util.VersionInfo;
+
+/**
+ * Builder for HTTP/2 {@link CloseableHttpAsyncClient} instances.
+ * <p>
+ * When a particular component is not explicitly set this class will
+ * use its default implementation.
+ *
+ * @since 5.0
+ */
+public class Http2AsyncClientBuilder {
+
+ private static class RequestInterceptorEntry {
+
+ enum Postion { FIRST, LAST }
+
+ final RequestInterceptorEntry.Postion postion;
+ final HttpRequestInterceptor interceptor;
+
+ private RequestInterceptorEntry(final RequestInterceptorEntry.Postion postion, final HttpRequestInterceptor interceptor) {
+ this.postion = postion;
+ this.interceptor = interceptor;
+ }
+ }
+
+ private static class ResponseInterceptorEntry {
+
+ enum Postion { FIRST, LAST }
+
+ final ResponseInterceptorEntry.Postion postion;
+ final HttpResponseInterceptor interceptor;
+
+ private ResponseInterceptorEntry(final ResponseInterceptorEntry.Postion postion, final HttpResponseInterceptor interceptor) {
+ this.postion = postion;
+ this.interceptor = interceptor;
+ }
+ }
+
+ private static class ExecInterceptorEntry {
+
+ enum Postion { BEFORE, AFTER, REPLACE, FIRST, LAST }
+
+ final ExecInterceptorEntry.Postion postion;
+ final String name;
+ final AsyncExecChainHandler interceptor;
+ final String existing;
+
+ private ExecInterceptorEntry(
+ final ExecInterceptorEntry.Postion postion,
+ final String name,
+ final AsyncExecChainHandler interceptor,
+ final String existing) {
+ this.postion = postion;
+ this.name = name;
+ this.interceptor = interceptor;
+ this.existing = existing;
+ }
+
+ }
+
+ private IOReactorConfig ioReactorConfig;
+ private H2Config h2Config;
+ private CharCodingConfig charCodingConfig;
+ private SchemePortResolver schemePortResolver;
+ private AuthenticationStrategy targetAuthStrategy;
+ private AuthenticationStrategy proxyAuthStrategy;
+
+ private LinkedList<RequestInterceptorEntry> requestInterceptors;
+ private LinkedList<ResponseInterceptorEntry> responseInterceptors;
+ private LinkedList<ExecInterceptorEntry> execInterceptors;
+
+ private HttpRoutePlanner routePlanner;
+ private RedirectStrategy redirectStrategy;
+ private HttpRequestRetryHandler retryHandler;
+
+ private Lookup<AuthSchemeProvider> authSchemeRegistry;
+ private Lookup<CookieSpecProvider> cookieSpecRegistry;
+ private CookieStore cookieStore;
+ private CredentialsProvider credentialsProvider;
+
+ private String userAgent;
+ private Collection<? extends Header> defaultHeaders;
+ private RequestConfig defaultRequestConfig;
+ private boolean evictIdleConnections;
+ private TimeValue maxIdleTime;
+
+ private boolean systemProperties;
+ private boolean automaticRetriesDisabled;
+ private boolean redirectHandlingDisabled;
+ private boolean cookieManagementDisabled;
+ private boolean authCachingDisabled;
+
+ private DnsResolver dnsResolver;
+ private TlsStrategy tlsStrategy;
+
+ private ThreadFactory threadFactory;
+
+ private List<Closeable> closeables;
+
+ public static Http2AsyncClientBuilder create() {
+ return new Http2AsyncClientBuilder();
+ }
+
+ protected Http2AsyncClientBuilder() {
+ super();
+ }
+
+ /**
+ * Sets {@link H2Config} configuration.
+ */
+ public final Http2AsyncClientBuilder setH2Config(final H2Config h2Config) {
+ this.h2Config = h2Config;
+ return this;
+ }
+
+ /**
+ * Sets {@link IOReactorConfig} configuration.
+ */
+ public final Http2AsyncClientBuilder setIOReactorConfig(final IOReactorConfig ioReactorConfig) {
+ this.ioReactorConfig = ioReactorConfig;
+ return this;
+ }
+
+ /**
+ * Sets {@link CharCodingConfig} configuration.
+ */
+ public final Http2AsyncClientBuilder setCharCodingConfig(final CharCodingConfig charCodingConfig) {
+ this.charCodingConfig = charCodingConfig;
+ return this;
+ }
+
+ /**
+ * Assigns {@link AuthenticationStrategy} instance for target
+ * host authentication.
+ */
+ public final Http2AsyncClientBuilder setTargetAuthenticationStrategy(
+ final AuthenticationStrategy targetAuthStrategy) {
+ this.targetAuthStrategy = targetAuthStrategy;
+ return this;
+ }
+
+ /**
+ * Assigns {@link AuthenticationStrategy} instance for proxy
+ * authentication.
+ */
+ public final Http2AsyncClientBuilder setProxyAuthenticationStrategy(
+ final AuthenticationStrategy proxyAuthStrategy) {
+ this.proxyAuthStrategy = proxyAuthStrategy;
+ return this;
+ }
+
+ /**
+ * Adds this protocol interceptor to the head of the protocol processing list.
+ */
+ public final Http2AsyncClientBuilder addRequestInterceptorFirst(final HttpResponseInterceptor interceptor) {
+ Args.notNull(interceptor, "Interceptor");
+ if (responseInterceptors == null) {
+ responseInterceptors = new LinkedList<>();
+ }
+ responseInterceptors.add(new ResponseInterceptorEntry(ResponseInterceptorEntry.Postion.FIRST, interceptor));
+ return this;
+ }
+
+ /**
+ * Adds this protocol interceptor to the tail of the protocol processing list.
+ */
+ public final Http2AsyncClientBuilder addResponseInterceptorLast(final HttpResponseInterceptor interceptor) {
+ Args.notNull(interceptor, "Interceptor");
+ if (responseInterceptors == null) {
+ responseInterceptors = new LinkedList<>();
+ }
+ responseInterceptors.add(new ResponseInterceptorEntry(ResponseInterceptorEntry.Postion.LAST, interceptor));
+ return this;
+ }
+
+ /**
+ * Adds this execution interceptor before an existing interceptor.
+ */
+ public final Http2AsyncClientBuilder addExecInterceptorBefore(final String existing, final String name, final AsyncExecChainHandler interceptor) {
+ Args.notBlank(existing, "Existing");
+ Args.notBlank(name, "Name");
+ Args.notNull(interceptor, "Interceptor");
+ if (execInterceptors == null) {
+ execInterceptors = new LinkedList<>();
+ }
+ execInterceptors.add(new ExecInterceptorEntry(ExecInterceptorEntry.Postion.BEFORE, name, interceptor, existing));
+ return this;
+ }
+
+ /**
+ * Adds this execution interceptor after interceptor with the given name.
+ */
+ public final Http2AsyncClientBuilder addExecInterceptorAfter(final String existing, final String name, final AsyncExecChainHandler interceptor) {
+ Args.notBlank(existing, "Existing");
+ Args.notBlank(name, "Name");
+ Args.notNull(interceptor, "Interceptor");
+ if (execInterceptors == null) {
+ execInterceptors = new LinkedList<>();
+ }
+ execInterceptors.add(new ExecInterceptorEntry(ExecInterceptorEntry.Postion.AFTER, name, interceptor, existing));
+ return this;
+ }
+
+ /**
+ * Replace an existing interceptor with the given name with new interceptor.
+ */
+ public final Http2AsyncClientBuilder replaceExecInterceptor(final String existing, final AsyncExecChainHandler interceptor) {
+ Args.notBlank(existing, "Existing");
+ Args.notNull(interceptor, "Interceptor");
+ if (execInterceptors == null) {
+ execInterceptors = new LinkedList<>();
+ }
+ execInterceptors.add(new ExecInterceptorEntry(ExecInterceptorEntry.Postion.REPLACE, existing, interceptor, existing));
+ return this;
+ }
+
+ /**
+ * Add an interceptor to the head of the processing list.
+ */
+ public final Http2AsyncClientBuilder addExecInterceptorFirst(final String name, final AsyncExecChainHandler interceptor) {
+ Args.notNull(name, "Name");
+ Args.notNull(interceptor, "Interceptor");
+ execInterceptors.add(new ExecInterceptorEntry(ExecInterceptorEntry.Postion.FIRST, name, interceptor, null));
+ return this;
+ }
+
+ /**
+ * Add an interceptor to the tail of the processing list.
+ */
+ public final Http2AsyncClientBuilder addExecInterceptorLast(final String name, final AsyncExecChainHandler interceptor) {
+ Args.notNull(name, "Name");
+ Args.notNull(interceptor, "Interceptor");
+ execInterceptors.add(new ExecInterceptorEntry(ExecInterceptorEntry.Postion.LAST, name, interceptor, null));
+ return this;
+ }
+
+ /**
+ * Adds this protocol interceptor to the head of the protocol processing list.
+ */
+ public final Http2AsyncClientBuilder addRequestInterceptorFirst(final HttpRequestInterceptor interceptor) {
+ Args.notNull(interceptor, "Interceptor");
+ if (requestInterceptors == null) {
+ requestInterceptors = new LinkedList<>();
+ }
+ requestInterceptors.add(new RequestInterceptorEntry(RequestInterceptorEntry.Postion.FIRST, interceptor));
+ return this;
+ }
+
+ /**
+ * Adds this protocol interceptor to the tail of the protocol processing list.
+ */
+ public final Http2AsyncClientBuilder addResponseInterceptorLast(final HttpRequestInterceptor interceptor) {
+ Args.notNull(interceptor, "Interceptor");
+ if (requestInterceptors == null) {
+ requestInterceptors = new LinkedList<>();
+ }
+ requestInterceptors.add(new RequestInterceptorEntry(RequestInterceptorEntry.Postion.LAST, interceptor));
+ return this;
+ }
+
+ /**
+ * Assigns {@link HttpRequestRetryHandler} instance.
+ * <p>
+ * Please note this value can be overridden by the {@link #disableAutomaticRetries()}
+ * method.
+ */
+ public final Http2AsyncClientBuilder setRetryHandler(final HttpRequestRetryHandler retryHandler) {
+ this.retryHandler = retryHandler;
+ return this;
+ }
+
+ /**
+ * Assigns {@link RedirectStrategy} instance.
+ * <p>
+ * Please note this value can be overridden by the {@link #disableRedirectHandling()}
+ * method.
+ * </p>
+ */
+ public Http2AsyncClientBuilder setRedirectStrategy(final RedirectStrategy redirectStrategy) {
+ this.redirectStrategy = redirectStrategy;
+ return this;
+ }
+
+ /**
+ * Assigns {@link SchemePortResolver} instance.
+ */
+ public final Http2AsyncClientBuilder setSchemePortResolver(final SchemePortResolver schemePortResolver) {
+ this.schemePortResolver = schemePortResolver;
+ return this;
+ }
+
+ /**
+ * Assigns {@link DnsResolver} instance.
+ */
+ public final Http2AsyncClientBuilder setDnsResolver(final DnsResolver dnsResolver) {
+ this.dnsResolver = dnsResolver;
+ return this;
+ }
+
+ /**
+ * Assigns {@link TlsStrategy} instance.
+ */
+ public final Http2AsyncClientBuilder setTlsStrategy(final TlsStrategy tlsStrategy) {
+ this.tlsStrategy = tlsStrategy;
+ return this;
+ }
+
+ /**
+ * Assigns {@link ThreadFactory} instance.
+ */
+ public final Http2AsyncClientBuilder setThreadFactory(final ThreadFactory threadFactory) {
+ this.threadFactory = threadFactory;
+ return this;
+ }
+
+ /**
+ * Assigns {@code User-Agent} value.
+ */
+ public final Http2AsyncClientBuilder setUserAgent(final String userAgent) {
+ this.userAgent = userAgent;
+ return this;
+ }
+
+ /**
+ * Assigns default request header values.
+ */
+ public final Http2AsyncClientBuilder setDefaultHeaders(final Collection<? extends Header> defaultHeaders) {
+ this.defaultHeaders = defaultHeaders;
+ return this;
+ }
+
+ /**
+ * Assigns {@link HttpRoutePlanner} instance.
+ */
+ public final Http2AsyncClientBuilder setRoutePlanner(final HttpRoutePlanner routePlanner) {
+ this.routePlanner = routePlanner;
+ return this;
+ }
+
+ /**
+ * Assigns default {@link CredentialsProvider} instance which will be used
+ * for request execution if not explicitly set in the client execution
+ * context.
+ */
+ public final Http2AsyncClientBuilder setDefaultCredentialsProvider(final CredentialsProvider credentialsProvider) {
+ this.credentialsProvider = credentialsProvider;
+ return this;
+ }
+
+ /**
+ * Assigns default {@link org.apache.hc.client5.http.auth.AuthScheme} registry which will
+ * be used for request execution if not explicitly set in the client execution
+ * context.
+ */
+ public final Http2AsyncClientBuilder setDefaultAuthSchemeRegistry(final Lookup<AuthSchemeProvider> authSchemeRegistry) {
+ this.authSchemeRegistry = authSchemeRegistry;
+ return this;
+ }
+
+ /**
+ * Assigns default {@link org.apache.hc.client5.http.cookie.CookieSpec} registry
+ * which will be used for request execution if not explicitly set in the client
+ * execution context.
+ */
+ public final Http2AsyncClientBuilder setDefaultCookieSpecRegistry(final Lookup<CookieSpecProvider> cookieSpecRegistry) {
+ this.cookieSpecRegistry = cookieSpecRegistry;
+ return this;
+ }
+
+ /**
+ * Assigns default {@link CookieStore} instance which will be used for
+ * request execution if not explicitly set in the client execution context.
+ */
+ public final Http2AsyncClientBuilder setDefaultCookieStore(final CookieStore cookieStore) {
+ this.cookieStore = cookieStore;
+ return this;
+ }
+
+ /**
+ * Assigns default {@link RequestConfig} instance which will be used
+ * for request execution if not explicitly set in the client execution
+ * context.
+ */
+ public final Http2AsyncClientBuilder setDefaultRequestConfig(final RequestConfig config) {
+ this.defaultRequestConfig = config;
+ return this;
+ }
+
+ /**
+ * Use system properties when creating and configuring default
+ * implementations.
+ */
+ public final Http2AsyncClientBuilder useSystemProperties() {
+ this.systemProperties = true;
+ return this;
+ }
+
+ /**
+ * Disables automatic redirect handling.
+ */
+ public final Http2AsyncClientBuilder disableRedirectHandling() {
+ redirectHandlingDisabled = true;
+ return this;
+ }
+
+ /**
+ * Disables automatic request recovery and re-execution.
+ */
+ public final Http2AsyncClientBuilder disableAutomaticRetries() {
+ automaticRetriesDisabled = true;
+ return this;
+ }
+
+ /**
+ * Disables state (cookie) management.
+ */
+ public final Http2AsyncClientBuilder disableCookieManagement() {
+ this.cookieManagementDisabled = true;
+ return this;
+ }
+
+ /**
+ * Disables authentication scheme caching.
+ */
+ public final Http2AsyncClientBuilder disableAuthCaching() {
+ this.authCachingDisabled = true;
+ return this;
+ }
+
+ /**
+ * Makes this instance of HttpClient proactively evict idle connections from the
+ * connection pool using a background thread.
+ * <p>
+ * One MUST explicitly close HttpClient with {@link CloseableHttpAsyncClient#close()}
+ * in order to stop and release the background thread.
+ * <p>
+ * Please note this method has no effect if the instance of HttpClient is configuted to
+ * use a shared connection manager.
+ *
+ * @param maxIdleTime maximum time persistent connections can stay idle while kept alive
+ * in the connection pool. Connections whose inactivity period exceeds this value will
+ * get closed and evicted from the pool.
+ */
+ public final Http2AsyncClientBuilder evictIdleConnections(final TimeValue maxIdleTime) {
+ this.evictIdleConnections = true;
+ this.maxIdleTime = maxIdleTime;
+ return this;
+ }
+
+ /**
+ * Request exec chain customization and extension.
+ * <p>
+ * For internal use.
+ */
+ @Internal
+ protected void customizeExecChain(final NamedElementChain<AsyncExecChainHandler> execChainDefinition) {
+ }
+
+ /**
+ * Adds to the list of {@link Closeable} resources to be managed by the client.
+ * <p>
+ * For internal use.
+ */
+ @Internal
+ protected void addCloseable(final Closeable closeable) {
+ if (closeable == null) {
+ return;
+ }
+ if (closeables == null) {
+ closeables = new ArrayList<>();
+ }
+ closeables.add(closeable);
+ }
+
+ public CloseableHttpAsyncClient build() {
+ final NamedElementChain<AsyncExecChainHandler> execChainDefinition = new NamedElementChain<>();
+ execChainDefinition.addLast(
+ new Http2AsyncMainClientExec(),
+ ChainElements.MAIN_TRANSPORT.name());
+
+ AuthenticationStrategy targetAuthStrategyCopy = this.targetAuthStrategy;
+ if (targetAuthStrategyCopy == null) {
+ targetAuthStrategyCopy = DefaultAuthenticationStrategy.INSTANCE;
+ }
+ AuthenticationStrategy proxyAuthStrategyCopy = this.proxyAuthStrategy;
+ if (proxyAuthStrategyCopy == null) {
+ proxyAuthStrategyCopy = DefaultAuthenticationStrategy.INSTANCE;
+ }
+
+ String userAgentCopy = this.userAgent;
+ if (userAgentCopy == null) {
+ if (systemProperties) {
+ userAgentCopy = getProperty("http.agent", null);
+ }
+ if (userAgentCopy == null) {
+ userAgentCopy = VersionInfo.getSoftwareInfo("Apache-HttpAsyncClient",
+ "org.apache.hc.client5", getClass());
+ }
+ }
+
+ execChainDefinition.addFirst(
+ new AsyncConnectExec(
+ new DefaultHttpProcessor(new RequestTargetHost(), new RequestUserAgent(userAgentCopy)),
+ proxyAuthStrategyCopy),
+ ChainElements.CONNECT.name());
+
+ final HttpProcessorBuilder b = HttpProcessorBuilder.create();
+ if (requestInterceptors != null) {
+ for (final RequestInterceptorEntry entry: requestInterceptors) {
+ if (entry.postion == RequestInterceptorEntry.Postion.FIRST) {
+ b.addFirst(entry.interceptor);
+ }
+ }
+ }
+ if (responseInterceptors != null) {
+ for (final ResponseInterceptorEntry entry: responseInterceptors) {
+ if (entry.postion == ResponseInterceptorEntry.Postion.FIRST) {
+ b.addFirst(entry.interceptor);
+ }
+ }
+ }
+ b.addAll(
+ new RequestDefaultHeaders(defaultHeaders),
+ new RequestUserAgent(userAgentCopy),
+ new RequestExpectContinue());
+ if (!cookieManagementDisabled) {
+ b.add(new RequestAddCookies());
+ }
+ if (!authCachingDisabled) {
+ b.add(new RequestAuthCache());
+ }
+ if (!cookieManagementDisabled) {
+ b.add(new ResponseProcessCookies());
+ }
+ if (requestInterceptors != null) {
+ for (final RequestInterceptorEntry entry: requestInterceptors) {
+ if (entry.postion == RequestInterceptorEntry.Postion.LAST) {
+ b.addFirst(entry.interceptor);
+ }
+ }
+ }
+ if (responseInterceptors != null) {
+ for (final ResponseInterceptorEntry entry: responseInterceptors) {
+ if (entry.postion == ResponseInterceptorEntry.Postion.LAST) {
+ b.addFirst(entry.interceptor);
+ }
+ }
+ }
+
+ final HttpProcessor httpProcessor = b.build();
+ execChainDefinition.addFirst(
+ new AsyncProtocolExec(httpProcessor, targetAuthStrategyCopy, proxyAuthStrategyCopy),
+ ChainElements.PROTOCOL.name());
+
+ // Add request retry executor, if not disabled
+ if (!automaticRetriesDisabled) {
+ HttpRequestRetryHandler retryHandlerCopy = this.retryHandler;
+ if (retryHandlerCopy == null) {
+ retryHandlerCopy = DefaultHttpRequestRetryHandler.INSTANCE;
+ }
+ execChainDefinition.addFirst(
+ new AsyncRetryExec(retryHandlerCopy),
+ ChainElements.RETRY_IO_ERROR.name());
+ }
+
+ HttpRoutePlanner routePlannerCopy = this.routePlanner;
+ if (routePlannerCopy == null) {
+ SchemePortResolver schemePortResolverCopy = this.schemePortResolver;
+ if (schemePortResolverCopy == null) {
+ schemePortResolverCopy = DefaultSchemePortResolver.INSTANCE;
+ }
+ routePlannerCopy = new DefaultRoutePlanner(schemePortResolverCopy);
+ }
+
+ // Add redirect executor, if not disabled
+ if (!redirectHandlingDisabled) {
+ RedirectStrategy redirectStrategyCopy = this.redirectStrategy;
+ if (redirectStrategyCopy == null) {
+ redirectStrategyCopy = DefaultRedirectStrategy.INSTANCE;
+ }
+ execChainDefinition.addFirst(
+ new AsyncRedirectExec(routePlannerCopy, redirectStrategyCopy),
+ ChainElements.REDIRECT.name());
+ }
+
+ final AsyncPushConsumerRegistry pushConsumerRegistry = new AsyncPushConsumerRegistry();
+ final IOEventHandlerFactory ioEventHandlerFactory = new Http2AsyncClientEventHandlerFactory(
+ new DefaultHttpProcessor(new H2RequestContent(), new H2RequestTargetHost(), new H2RequestConnControl()),
+ new HandlerFactory<AsyncPushConsumer>() {
+
+ @Override
+ public AsyncPushConsumer create(final HttpRequest request, final HttpContext context) throws HttpException {
+ return pushConsumerRegistry.get(request);
+ }
+
+ },
+ h2Config != null ? h2Config : H2Config.DEFAULT,
+ charCodingConfig != null ? charCodingConfig : CharCodingConfig.DEFAULT);
+ final DefaultConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(
+ ioEventHandlerFactory,
+ ioReactorConfig != null ? ioReactorConfig : IOReactorConfig.DEFAULT,
+ threadFactory != null ? threadFactory : new DefaultThreadFactory("httpclient-dispatch", true),
+ null,
+ null,
+ new Callback<IOSession>() {
+
+ @Override
+ public void execute(final IOSession ioSession) {
+ ioSession.addFirst(new ShutdownCommand(ShutdownType.GRACEFUL));
+ }
+
+ });
+
+ if (execInterceptors != null) {
+ for (final ExecInterceptorEntry entry: execInterceptors) {
+ switch (entry.postion) {
+ case AFTER:
+ execChainDefinition.addAfter(entry.existing, entry.interceptor, entry.name);
+ break;
+ case BEFORE:
+ execChainDefinition.addBefore(entry.existing, entry.interceptor, entry.name);
+ break;
+ case FIRST:
+ execChainDefinition.addFirst(entry.interceptor, entry.name);
+ break;
+ case LAST:
+ execChainDefinition.addLast(entry.interceptor, entry.name);
+ break;
+ }
+ }
+ }
+
+ customizeExecChain(execChainDefinition);
+
+ NamedElementChain<AsyncExecChainHandler>.Node current = execChainDefinition.getLast();
+ AsyncExecChainElement execChain = null;
+ while (current != null) {
+ execChain = new AsyncExecChainElement(current.getValue(), execChain);
+ current = current.getPrevious();
+ }
+
+ Lookup<AuthSchemeProvider> authSchemeRegistryCopy = this.authSchemeRegistry;
+ if (authSchemeRegistryCopy == null) {
+ authSchemeRegistryCopy = RegistryBuilder.<AuthSchemeProvider>create()
+ .register(AuthSchemes.BASIC, new BasicSchemeFactory())
+ .register(AuthSchemes.DIGEST, new DigestSchemeFactory())
+ .register(AuthSchemes.CREDSSP, new CredSspSchemeFactory())
+ .register(AuthSchemes.NTLM, new NTLMSchemeFactory())
+ .register(AuthSchemes.SPNEGO, new SPNegoSchemeFactory(KerberosConfig.DEFAULT, SystemDefaultDnsResolver.INSTANCE))
+ .register(AuthSchemes.KERBEROS, new KerberosSchemeFactory(KerberosConfig.DEFAULT, SystemDefaultDnsResolver.INSTANCE))
+ .build();
+ }
+ Lookup<CookieSpecProvider> cookieSpecRegistryCopy = this.cookieSpecRegistry;
+ if (cookieSpecRegistryCopy == null) {
+ cookieSpecRegistryCopy = CookieSpecRegistries.createDefault();
+ }
+
+ CookieStore cookieStoreCopy = this.cookieStore;
+ if (cookieStoreCopy == null) {
+ cookieStoreCopy = new BasicCookieStore();
+ }
+
+ CredentialsProvider credentialsProviderCopy = this.credentialsProvider;
+ if (credentialsProviderCopy == null) {
+ if (systemProperties) {
+ credentialsProviderCopy = new SystemDefaultCredentialsProvider();
+ } else {
+ credentialsProviderCopy = new BasicCredentialsProvider();
+ }
+ }
+
+ TlsStrategy tlsStrategyCopy = this.tlsStrategy;
+ if (tlsStrategyCopy == null) {
+ if (systemProperties) {
+ tlsStrategyCopy = H2TlsStrategy.getSystemDefault();
+ } else {
+ tlsStrategyCopy = H2TlsStrategy.getDefault();
+ }
+ }
+
+ final MultuhomeConnectionInitiator connectionInitiator = new MultuhomeConnectionInitiator(ioReactor, dnsResolver);
+ final H2ConnPool connPool = new H2ConnPool(connectionInitiator, new Resolver<HttpHost, InetSocketAddress>() {
+
+ @Override
+ public InetSocketAddress resolve(final HttpHost host) {
+ return null;
+ }
+
+ }, tlsStrategyCopy);
+
+ List<Closeable> closeablesCopy = closeables != null ? new ArrayList<>(closeables) : null;
+ if (closeablesCopy == null) {
+ closeablesCopy = new ArrayList<>(1);
+ }
+ if (evictIdleConnections) {
+ final IdleConnectionEvictor connectionEvictor = new IdleConnectionEvictor(connPool,
+ maxIdleTime != null ? maxIdleTime : TimeValue.ofSeconds(30L));
+ closeablesCopy.add(new Closeable() {
+
+ @Override
+ public void close() throws IOException {
+ connectionEvictor.shutdown();
+ }
+
+ });
+ connectionEvictor.start();
+ }
+ closeablesCopy.add(connPool);
+
+ return new InternalHttp2AsyncClient(
+ ioReactor,
+ execChain,
+ pushConsumerRegistry,
+ threadFactory != null ? threadFactory : new DefaultThreadFactory("httpclient-main", true),
+ connPool,
+ routePlannerCopy,
+ cookieSpecRegistryCopy,
+ authSchemeRegistryCopy,
+ cookieStoreCopy,
+ credentialsProviderCopy,
+ defaultRequestConfig,
+ closeablesCopy);
+ }
+
+ private static String getProperty(final String key, final String defaultValue) {
+ return AccessController.doPrivileged(new PrivilegedAction<String>() {
+ @Override
+ public String run() {
+ return System.getProperty(key, defaultValue);
+ }
+ });
+ }
+
+ static class IdleConnectionEvictor implements Closeable {
+
+ private final Thread thread;
+
+ public IdleConnectionEvictor(final H2ConnPool connPool, final TimeValue maxIdleTime) {
+ this.thread = new DefaultThreadFactory("idle-connection-evictor", true).newThread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ Thread.sleep(maxIdleTime.toMillis());
+ connPool.closeIdle(maxIdleTime);
+ }
+ } catch (final InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ } catch (final Exception ex) {
+ }
+
+ }
+ });
+ }
+
+ public void start() {
+ thread.start();
+ }
+
+ public void shutdown() {
+ thread.interrupt();
+ }
+
+ @Override
+ public void close() throws IOException {
+ shutdown();
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientEventHandlerFactory.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientEventHandlerFactory.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientEventHandlerFactory.java
new file mode 100644
index 0000000..52f399c
--- /dev/null
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncClientEventHandlerFactory.java
@@ -0,0 +1,194 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.client5.http.impl.async;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hc.client5.http.impl.ConnPoolSupport;
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpConnection;
+import org.apache.hc.core5.http.config.CharCodingConfig;
+import org.apache.hc.core5.http.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http.nio.HandlerFactory;
+import org.apache.hc.core5.http.protocol.HttpProcessor;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.frame.FramePrinter;
+import org.apache.hc.core5.http2.frame.RawFrame;
+import org.apache.hc.core5.http2.impl.nio.ClientHttp2StreamMultiplexerFactory;
+import org.apache.hc.core5.http2.impl.nio.Http2OnlyClientProtocolNegotiator;
+import org.apache.hc.core5.http2.impl.nio.Http2StreamListener;
+import org.apache.hc.core5.reactor.IOEventHandler;
+import org.apache.hc.core5.reactor.IOEventHandlerFactory;
+import org.apache.hc.core5.reactor.TlsCapableIOSession;
+import org.apache.hc.core5.util.Args;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * @since 5.0
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE)
+class Http2AsyncClientEventHandlerFactory implements IOEventHandlerFactory {
+
+ private final Logger wireLog = LogManager.getLogger("org.apache.hc.client5.http.wire");
+ private final Logger headerLog = LogManager.getLogger("org.apache.hc.client5.http.headers");
+ private final Logger frameLog = LogManager.getLogger("org.apache.hc.client5.http2.frame");
+ private final Logger framePayloadLog = LogManager.getLogger("org.apache.hc.client5.http2.frame.payload");
+ private final Logger flowCtrlLog = LogManager.getLogger("org.apache.hc.client5.http2.flow");
+
+ private final HttpProcessor httpProcessor;
+ private final HandlerFactory<AsyncPushConsumer> exchangeHandlerFactory;
+ private final H2Config h2Config;
+ private final CharCodingConfig charCodingConfig;
+
+ Http2AsyncClientEventHandlerFactory(
+ final HttpProcessor httpProcessor,
+ final HandlerFactory<AsyncPushConsumer> exchangeHandlerFactory,
+ final H2Config h2Config,
+ final CharCodingConfig charCodingConfig) {
+ this.httpProcessor = Args.notNull(httpProcessor, "HTTP processor");
+ this.exchangeHandlerFactory = exchangeHandlerFactory;
+ this.h2Config = h2Config != null ? h2Config : H2Config.DEFAULT;
+ this.charCodingConfig = charCodingConfig != null ? charCodingConfig : CharCodingConfig.DEFAULT;
+ }
+
+ @Override
+ public IOEventHandler createHandler(final TlsCapableIOSession ioSession, final Object attachment) {
+ final Logger sessionLog = LogManager.getLogger(ioSession.getClass());
+ if (sessionLog.isDebugEnabled()
+ || wireLog.isDebugEnabled()
+ || headerLog.isDebugEnabled()
+ || frameLog.isDebugEnabled()
+ || framePayloadLog.isDebugEnabled()
+ || flowCtrlLog.isDebugEnabled()) {
+ final String id = ConnPoolSupport.getId(ioSession);
+ final ClientHttp2StreamMultiplexerFactory http2StreamHandlerFactory = new ClientHttp2StreamMultiplexerFactory(
+ httpProcessor,
+ exchangeHandlerFactory,
+ h2Config,
+ charCodingConfig,
+ new Http2StreamListener() {
+
+ final FramePrinter framePrinter = new FramePrinter();
+
+ private void logFrameInfo(final String prefix, final RawFrame frame) {
+ try {
+ final LogAppendable logAppendable = new LogAppendable(frameLog, prefix);
+ framePrinter.printFrameInfo(frame, logAppendable);
+ logAppendable.flush();
+ } catch (final IOException ignore) {
+ }
+ }
+
+ private void logFramePayload(final String prefix, final RawFrame frame) {
+ try {
+ final LogAppendable logAppendable = new LogAppendable(framePayloadLog, prefix);
+ framePrinter.printPayload(frame, logAppendable);
+ logAppendable.flush();
+ } catch (final IOException ignore) {
+ }
+ }
+
+ private void logFlowControl(final String prefix, final int streamId, final int delta, final int actualSize) {
+ final StringBuilder buffer = new StringBuilder();
+ buffer.append(prefix).append(" stream ").append(streamId).append(" flow control " )
+ .append(delta).append(" -> ")
+ .append(actualSize);
+ flowCtrlLog.debug(buffer.toString());
+ }
+
+ @Override
+ public void onHeaderInput(final HttpConnection connection, final int streamId, final List<? extends Header> headers) {
+ if (headerLog.isDebugEnabled()) {
+ for (int i = 0; i < headers.size(); i++) {
+ headerLog.debug(id + " << " + headers.get(i));
+ }
+ }
+ }
+
+ @Override
+ public void onHeaderOutput(final HttpConnection connection, final int streamId, final List<? extends Header> headers) {
+ if (headerLog.isDebugEnabled()) {
+ for (int i = 0; i < headers.size(); i++) {
+ headerLog.debug(id + " >> " + headers.get(i));
+ }
+ }
+ }
+
+ @Override
+ public void onFrameInput(final HttpConnection connection, final int streamId, final RawFrame frame) {
+ if (frameLog.isDebugEnabled()) {
+ logFrameInfo(id + " <<", frame);
+ }
+ if (framePayloadLog.isDebugEnabled()) {
+ logFramePayload(id + " <<", frame);
+ }
+ }
+
+ @Override
+ public void onFrameOutput(final HttpConnection connection, final int streamId, final RawFrame frame) {
+ if (frameLog.isDebugEnabled()) {
+ logFrameInfo(id + " >>", frame);
+ }
+ if (framePayloadLog.isDebugEnabled()) {
+ logFramePayload(id + " >>", frame);
+ }
+ }
+
+ @Override
+ public void onInputFlowControl(final HttpConnection connection, final int streamId, final int delta, final int actualSize) {
+ if (flowCtrlLog.isDebugEnabled()) {
+ logFlowControl(id + " <<", streamId, delta, actualSize);
+ }
+ }
+
+ @Override
+ public void onOutputFlowControl(final HttpConnection connection, final int streamId, final int delta, final int actualSize) {
+ if (flowCtrlLog.isDebugEnabled()) {
+ logFlowControl(id + " >>", streamId, delta, actualSize);
+ }
+ }
+
+ });
+ final LoggingIOSession loggingIOSession = new LoggingIOSession(ioSession, id, sessionLog, wireLog);
+ return new Http2OnlyClientProtocolNegotiator(loggingIOSession, http2StreamHandlerFactory);
+ } else {
+ final ClientHttp2StreamMultiplexerFactory http2StreamHandlerFactory = new ClientHttp2StreamMultiplexerFactory(
+ httpProcessor,
+ exchangeHandlerFactory,
+ h2Config,
+ charCodingConfig,
+ null);
+ return new Http2OnlyClientProtocolNegotiator(ioSession, http2StreamHandlerFactory);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncMainClientExec.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncMainClientExec.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncMainClientExec.java
new file mode 100644
index 0000000..a954a56
--- /dev/null
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/Http2AsyncMainClientExec.java
@@ -0,0 +1,170 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.impl.async;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hc.client5.http.HttpRoute;
+import org.apache.hc.client5.http.async.AsyncExecCallback;
+import org.apache.hc.client5.http.async.AsyncExecChain;
+import org.apache.hc.client5.http.async.AsyncExecChainHandler;
+import org.apache.hc.client5.http.async.AsyncExecRuntime;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.message.RequestLine;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
+import org.apache.hc.core5.http.nio.AsyncDataConsumer;
+import org.apache.hc.core5.http.nio.AsyncEntityProducer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
+import org.apache.hc.core5.http.nio.RequestChannel;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+class Http2AsyncMainClientExec implements AsyncExecChainHandler {
+
+ private final Logger log = LogManager.getLogger(getClass());
+
+ @Override
+ public void execute(
+ final HttpRequest request,
+ final AsyncEntityProducer entityProducer,
+ final AsyncExecChain.Scope scope,
+ final AsyncExecChain chain,
+ final AsyncExecCallback asyncExecCallback) throws HttpException, IOException {
+ final String exchangeId = scope.exchangeId;
+ final HttpRoute route = scope.route;
+ final HttpClientContext clientContext = scope.clientContext;
+ final AsyncExecRuntime execRuntime = scope.execRuntime;
+
+ if (log.isDebugEnabled()) {
+ log.debug(exchangeId + ": executing " + new RequestLine(request));
+ }
+
+ final AsyncClientExchangeHandler internalExchangeHandler = new AsyncClientExchangeHandler() {
+
+ private final AtomicReference<AsyncDataConsumer> entityConsumerRef = new AtomicReference<>(null);
+
+ @Override
+ public void releaseResources() {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
+ if (entityConsumer != null) {
+ entityConsumer.releaseResources();
+ }
+ }
+
+ @Override
+ public void failed(final Exception cause) {
+ execRuntime.markConnectionNonReusable();
+ asyncExecCallback.failed(cause);
+ }
+
+ @Override
+ public void cancel() {
+ failed(new InterruptedIOException());
+ }
+
+ @Override
+ public void produceRequest(final RequestChannel channel) throws HttpException, IOException {
+ channel.sendRequest(request, entityProducer);
+ }
+
+ @Override
+ public int available() {
+ return entityProducer.available();
+ }
+
+ @Override
+ public void produce(final DataStreamChannel channel) throws IOException {
+ entityProducer.produce(channel);
+ }
+
+ @Override
+ public void consumeInformation(final HttpResponse response) throws HttpException, IOException {
+ }
+
+ @Override
+ public void consumeResponse(final HttpResponse response, final EntityDetails entityDetails) throws HttpException, IOException {
+ entityConsumerRef.set(asyncExecCallback.handleResponse(response, entityDetails));
+ if (entityDetails == null) {
+ execRuntime.validateConnection();
+ asyncExecCallback.completed();
+ }
+ }
+
+ @Override
+ public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
+ if (entityConsumer != null) {
+ entityConsumer.updateCapacity(capacityChannel);
+ } else {
+ capacityChannel.update(Integer.MAX_VALUE);
+ }
+ }
+
+ @Override
+ public int consume(final ByteBuffer src) throws IOException {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
+ if (entityConsumer != null) {
+ return entityConsumer.consume(src);
+ } else {
+ return Integer.MAX_VALUE;
+ }
+ }
+
+ @Override
+ public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
+ if (entityConsumer != null) {
+ entityConsumer.streamEnd(trailers);
+ } else {
+ execRuntime.validateConnection();
+ }
+ asyncExecCallback.completed();
+ }
+
+ };
+
+ if (log.isDebugEnabled()) {
+ execRuntime.execute(
+ new LoggingAsyncClientExchangeHandler(log, exchangeId, internalExchangeHandler),
+ clientContext);
+ } else {
+ execRuntime.execute(
+ internalExchangeHandler, clientContext);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientBuilder.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientBuilder.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientBuilder.java
index 87ce294..91c28ae 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientBuilder.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientBuilder.java
@@ -736,7 +736,7 @@ public class HttpAsyncClientBuilder {
final NamedElementChain<AsyncExecChainHandler> execChainDefinition = new NamedElementChain<>();
execChainDefinition.addLast(
- new AsyncMainClientExec(keepAliveStrategyCopy, userTokenHandlerCopy),
+ new HttpAsyncMainClientExec(keepAliveStrategyCopy, userTokenHandlerCopy),
ChainElements.MAIN_TRANSPORT.name());
AuthenticationStrategy targetAuthStrategyCopy = this.targetAuthStrategy;
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientEventHandlerFactory.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientEventHandlerFactory.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientEventHandlerFactory.java
index 95d9267..4a97d67 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientEventHandlerFactory.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientEventHandlerFactory.java
@@ -71,7 +71,7 @@ import org.apache.logging.log4j.Logger;
* @since 5.0
*/
@Contract(threading = ThreadingBehavior.IMMUTABLE)
-public class HttpAsyncClientEventHandlerFactory implements IOEventHandlerFactory {
+class HttpAsyncClientEventHandlerFactory implements IOEventHandlerFactory {
private final Logger streamLog = LogManager.getLogger(InternalHttpAsyncClient.class);
private final Logger wireLog = LogManager.getLogger("org.apache.hc.client5.http.wire");
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClients.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClients.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClients.java
index 56a59c3..dab74fd 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClients.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClients.java
@@ -88,6 +88,31 @@ public class HttpAsyncClients {
return HttpAsyncClientBuilder.create().useSystemProperties().build();
}
+ /**
+ * Creates builder object for construction of custom HTTP/2
+ * {@link CloseableHttpAsyncClient} instances optimized for HTTP/2 protocol
+ * and message multiplexing
+ */
+ public static Http2AsyncClientBuilder customHttp2() {
+ return Http2AsyncClientBuilder.create();
+ }
+
+ /**
+ * Creates HTTP/2 {@link CloseableHttpAsyncClient} instance with default configuration
+ * optimized for HTTP/2 protocol and message multiplexing.
+ */
+ public static CloseableHttpAsyncClient createHttp2Default() {
+ return Http2AsyncClientBuilder.create().build();
+ }
+
+ /**
+ * Creates HTTP/2 {@link CloseableHttpAsyncClient} instance with default configuration and
+ * system properties optimized for HTTP/2 protocol and message multiplexing.
+ */
+ public static CloseableHttpAsyncClient createHttp2System() {
+ return Http2AsyncClientBuilder.create().useSystemProperties().build();
+ }
+
private static HttpProcessor createMinimalProtocolProcessor() {
return new DefaultHttpProcessor(
new H2RequestContent(),
@@ -220,7 +245,7 @@ public class HttpAsyncClients {
final TlsStrategy tlsStrategy) {
final AsyncPushConsumerRegistry pushConsumerRegistry = new AsyncPushConsumerRegistry();
return createMinimalHttp2AsyncClientImpl(
- new HttpAsyncClientEventHandlerFactory(
+ new Http2AsyncClientEventHandlerFactory(
createMinimalProtocolProcessor(),
new HandlerFactory<AsyncPushConsumer>() {
@@ -230,11 +255,8 @@ public class HttpAsyncClients {
}
},
- HttpVersionPolicy.FORCE_HTTP_2,
h2Config,
- null,
- CharCodingConfig.DEFAULT,
- DefaultConnectionReuseStrategy.INSTANCE),
+ CharCodingConfig.DEFAULT),
pushConsumerRegistry,
ioReactorConfig,
dnsResolver,
http://git-wip-us.apache.org/repos/asf/httpcomponents-client/blob/6228a736/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncMainClientExec.java
----------------------------------------------------------------------
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncMainClientExec.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncMainClientExec.java
new file mode 100644
index 0000000..012803c
--- /dev/null
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncMainClientExec.java
@@ -0,0 +1,229 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.impl.async;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hc.client5.http.ConnectionKeepAliveStrategy;
+import org.apache.hc.client5.http.HttpRoute;
+import org.apache.hc.client5.http.UserTokenHandler;
+import org.apache.hc.client5.http.async.AsyncExecCallback;
+import org.apache.hc.client5.http.async.AsyncExecChain;
+import org.apache.hc.client5.http.async.AsyncExecChainHandler;
+import org.apache.hc.client5.http.async.AsyncExecRuntime;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.hc.core5.http.message.RequestLine;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
+import org.apache.hc.core5.http.nio.AsyncDataConsumer;
+import org.apache.hc.core5.http.nio.AsyncEntityProducer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
+import org.apache.hc.core5.http.nio.RequestChannel;
+import org.apache.hc.core5.util.TimeValue;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+class HttpAsyncMainClientExec implements AsyncExecChainHandler {
+
+ private final Logger log = LogManager.getLogger(getClass());
+
+ private final ConnectionKeepAliveStrategy keepAliveStrategy;
+ private final UserTokenHandler userTokenHandler;
+
+ HttpAsyncMainClientExec(final ConnectionKeepAliveStrategy keepAliveStrategy, final UserTokenHandler userTokenHandler) {
+ this.keepAliveStrategy = keepAliveStrategy;
+ this.userTokenHandler = userTokenHandler;
+ }
+
+ @Override
+ public void execute(
+ final HttpRequest request,
+ final AsyncEntityProducer entityProducer,
+ final AsyncExecChain.Scope scope,
+ final AsyncExecChain chain,
+ final AsyncExecCallback asyncExecCallback) throws HttpException, IOException {
+ final String exchangeId = scope.exchangeId;
+ final HttpRoute route = scope.route;
+ final HttpClientContext clientContext = scope.clientContext;
+ final AsyncExecRuntime execRuntime = scope.execRuntime;
+
+ if (log.isDebugEnabled()) {
+ log.debug(exchangeId + ": executing " + new RequestLine(request));
+ }
+
+ final AtomicInteger messageCountDown = new AtomicInteger(2);
+ final AsyncClientExchangeHandler internalExchangeHandler = new AsyncClientExchangeHandler() {
+
+ private final AtomicReference<AsyncDataConsumer> entityConsumerRef = new AtomicReference<>(null);
+
+ @Override
+ public void releaseResources() {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
+ if (entityConsumer != null) {
+ entityConsumer.releaseResources();
+ }
+ }
+
+ @Override
+ public void failed(final Exception cause) {
+ execRuntime.markConnectionNonReusable();
+ asyncExecCallback.failed(cause);
+ }
+
+ @Override
+ public void cancel() {
+ failed(new InterruptedIOException());
+ }
+
+ @Override
+ public void produceRequest(final RequestChannel channel) throws HttpException, IOException {
+ channel.sendRequest(request, entityProducer);
+ if (entityProducer == null) {
+ messageCountDown.decrementAndGet();
+ }
+ }
+
+ @Override
+ public int available() {
+ return entityProducer.available();
+ }
+
+ @Override
+ public void produce(final DataStreamChannel channel) throws IOException {
+ entityProducer.produce(new DataStreamChannel() {
+
+ @Override
+ public void requestOutput() {
+ channel.requestOutput();
+ }
+
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
+ return channel.write(src);
+ }
+
+ @Override
+ public void endStream(final List<? extends Header> trailers) throws IOException {
+ channel.endStream(trailers);
+ if (messageCountDown.decrementAndGet() <= 0) {
+ asyncExecCallback.completed();
+ }
+ }
+
+ @Override
+ public void endStream() throws IOException {
+ channel.endStream();
+ if (messageCountDown.decrementAndGet() <= 0) {
+ asyncExecCallback.completed();
+ }
+ }
+
+ });
+ }
+
+ @Override
+ public void consumeInformation(final HttpResponse response) throws HttpException, IOException {
+ }
+
+ @Override
+ public void consumeResponse(final HttpResponse response, final EntityDetails entityDetails) throws HttpException, IOException {
+ entityConsumerRef.set(asyncExecCallback.handleResponse(response, entityDetails));
+ if (response.getCode() >= HttpStatus.SC_CLIENT_ERROR) {
+ messageCountDown.decrementAndGet();
+ }
+ final TimeValue keepAliveDuration = keepAliveStrategy.getKeepAliveDuration(response, clientContext);
+ Object userToken = clientContext.getUserToken();
+ if (userToken == null) {
+ userToken = userTokenHandler.getUserToken(route, clientContext);
+ clientContext.setAttribute(HttpClientContext.USER_TOKEN, userToken);
+ }
+ execRuntime.markConnectionReusable(userToken, keepAliveDuration);
+ if (entityDetails == null) {
+ execRuntime.validateConnection();
+ if (messageCountDown.decrementAndGet() <= 0) {
+ asyncExecCallback.completed();
+ }
+ }
+ }
+
+ @Override
+ public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
+ if (entityConsumer != null) {
+ entityConsumer.updateCapacity(capacityChannel);
+ } else {
+ capacityChannel.update(Integer.MAX_VALUE);
+ }
+ }
+
+ @Override
+ public int consume(final ByteBuffer src) throws IOException {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
+ if (entityConsumer != null) {
+ return entityConsumer.consume(src);
+ } else {
+ return Integer.MAX_VALUE;
+ }
+ }
+
+ @Override
+ public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
+ final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
+ if (entityConsumer != null) {
+ entityConsumer.streamEnd(trailers);
+ } else {
+ execRuntime.validateConnection();
+ }
+ if (messageCountDown.decrementAndGet() <= 0) {
+ asyncExecCallback.completed();
+ }
+ }
+
+ };
+
+ if (log.isDebugEnabled()) {
+ execRuntime.execute(
+ new LoggingAsyncClientExchangeHandler(log, exchangeId, internalExchangeHandler),
+ clientContext);
+ } else {
+ execRuntime.execute(
+ internalExchangeHandler, clientContext);
+ }
+ }
+
+}