DefaultAsyncClientConnectionOperator.java
/*
* ====================================================================
* 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.nio;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.Future;
import org.apache.hc.client5.http.DnsResolver;
import org.apache.hc.client5.http.SchemePortResolver;
import org.apache.hc.client5.http.UnsupportedSchemeException;
import org.apache.hc.client5.http.config.TlsConfig;
import org.apache.hc.client5.http.impl.ConnPoolSupport;
import org.apache.hc.client5.http.impl.DefaultSchemePortResolver;
import org.apache.hc.client5.http.nio.AsyncClientConnectionOperator;
import org.apache.hc.client5.http.nio.ManagedAsyncClientConnection;
import org.apache.hc.client5.http.routing.RoutingSupport;
import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.ComplexFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.URIScheme;
import org.apache.hc.core5.http.config.Lookup;
import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.net.NamedEndpoint;
import org.apache.hc.core5.reactor.ConnectionInitiator;
import org.apache.hc.core5.reactor.IOSession;
import org.apache.hc.core5.reactor.ssl.TransportSecurityLayer;
import org.apache.hc.core5.util.Args;
import org.apache.hc.core5.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@Internal
public class DefaultAsyncClientConnectionOperator implements AsyncClientConnectionOperator {
private static final Logger LOG = LoggerFactory.getLogger(DefaultAsyncClientConnectionOperator.class);
private final SchemePortResolver schemePortResolver;
private final MultihomeIOSessionRequester sessionRequester;
private final Lookup<TlsStrategy> tlsStrategyLookup;
DefaultAsyncClientConnectionOperator(
final Lookup<TlsStrategy> tlsStrategyLookup,
final SchemePortResolver schemePortResolver,
final DnsResolver dnsResolver) {
this.tlsStrategyLookup = Args.notNull(tlsStrategyLookup, "TLS strategy lookup");
this.schemePortResolver = schemePortResolver != null ? schemePortResolver : DefaultSchemePortResolver.INSTANCE;
this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver);
}
@Override
public Future<ManagedAsyncClientConnection> connect(
final ConnectionInitiator connectionInitiator,
final HttpHost host,
final SocketAddress localAddress,
final Timeout connectTimeout,
final Object attachment,
final FutureCallback<ManagedAsyncClientConnection> callback) {
return connect(connectionInitiator, host, null, localAddress, connectTimeout,
attachment, null, callback);
}
@Override
public Future<ManagedAsyncClientConnection> connect(
final ConnectionInitiator connectionInitiator,
final HttpHost endpointHost,
final NamedEndpoint endpointName,
final SocketAddress localAddress,
final Timeout connectTimeout,
final Object attachment,
final HttpContext context,
final FutureCallback<ManagedAsyncClientConnection> callback) {
Args.notNull(connectionInitiator, "Connection initiator");
Args.notNull(endpointHost, "Host");
final ComplexFuture<ManagedAsyncClientConnection> future = new ComplexFuture<>(callback);
final HttpHost remoteEndpoint = RoutingSupport.normalize(endpointHost, schemePortResolver);
final InetAddress remoteAddress = endpointHost.getAddress();
final TlsConfig tlsConfig = attachment instanceof TlsConfig ? (TlsConfig) attachment : TlsConfig.DEFAULT;
onBeforeSocketConnect(context, endpointHost);
if (LOG.isDebugEnabled()) {
LOG.debug("{} connecting {}->{} ({})", endpointHost, localAddress, remoteAddress, connectTimeout);
}
final Future<IOSession> sessionFuture = sessionRequester.connect(
connectionInitiator,
remoteEndpoint,
remoteAddress != null ? new InetSocketAddress(remoteAddress, remoteEndpoint.getPort()) : null,
localAddress,
connectTimeout,
tlsConfig.getHttpVersionPolicy(),
new FutureCallback<IOSession>() {
@Override
public void completed(final IOSession session) {
final DefaultManagedAsyncClientConnection connection = new DefaultManagedAsyncClientConnection(session);
onAfterSocketConnect(context, endpointHost);
if (LOG.isDebugEnabled()) {
LOG.debug("{} {} connected {}->{}", ConnPoolSupport.getId(connection), endpointHost,
connection.getLocalAddress(), connection.getRemoteAddress());
}
final TlsStrategy tlsStrategy = tlsStrategyLookup != null ? tlsStrategyLookup.lookup(endpointHost.getSchemeName()) : null;
if (tlsStrategy != null) {
try {
final Timeout socketTimeout = connection.getSocketTimeout();
final Timeout handshakeTimeout = tlsConfig.getHandshakeTimeout();
final NamedEndpoint tlsName = endpointName != null ? endpointName : endpointHost;
onBeforeTlsHandshake(context, endpointHost);
if (LOG.isDebugEnabled()) {
LOG.debug("{} {} upgrading to TLS", ConnPoolSupport.getId(connection), tlsName);
}
tlsStrategy.upgrade(
connection,
tlsName,
attachment,
handshakeTimeout != null ? handshakeTimeout : connectTimeout,
new FutureContribution<TransportSecurityLayer>(future) {
@Override
public void completed(final TransportSecurityLayer transportSecurityLayer) {
connection.setSocketTimeout(socketTimeout);
future.completed(connection);
onAfterTlsHandshake(context, endpointHost);
if (LOG.isDebugEnabled()) {
LOG.debug("{} {} upgraded to TLS", ConnPoolSupport.getId(connection), tlsName);
}
}
});
} catch (final Exception ex) {
future.failed(ex);
}
} else {
future.completed(connection);
}
}
@Override
public void failed(final Exception ex) {
future.failed(ex);
}
@Override
public void cancelled() {
future.cancel();
}
});
future.setDependency(sessionFuture);
return future;
}
@Override
public void upgrade(
final ManagedAsyncClientConnection connection,
final HttpHost host,
final Object attachment) {
upgrade(connection, host, null, attachment, null, null);
}
@Override
public void upgrade(
final ManagedAsyncClientConnection connection,
final HttpHost endpointHost,
final NamedEndpoint endpointName,
final Object attachment,
final HttpContext context,
final FutureCallback<ManagedAsyncClientConnection> callback) {
final String newProtocol = URIScheme.HTTP.same(endpointHost.getSchemeName()) ? URIScheme.HTTPS.id : endpointHost.getSchemeName();
final TlsStrategy tlsStrategy = tlsStrategyLookup != null ? tlsStrategyLookup.lookup(newProtocol) : null;
if (tlsStrategy != null) {
final NamedEndpoint tlsName = endpointName != null ? endpointName : endpointHost;
if (LOG.isDebugEnabled()) {
LOG.debug("{} {} upgrading to TLS", ConnPoolSupport.getId(connection), tlsName);
}
tlsStrategy.upgrade(
connection,
tlsName,
attachment,
null,
new CallbackContribution<TransportSecurityLayer>(callback) {
@Override
public void completed(final TransportSecurityLayer transportSecurityLayer) {
if (callback != null) {
callback.completed(connection);
}
}
});
} else {
callback.failed(new UnsupportedSchemeException(newProtocol + " protocol is not supported"));
}
}
protected void onBeforeSocketConnect(final HttpContext httpContext, final HttpHost endpointHost) {
}
protected void onAfterSocketConnect(final HttpContext httpContext, final HttpHost endpointHost) {
}
protected void onBeforeTlsHandshake(final HttpContext httpContext, final HttpHost endpointHost) {
}
protected void onAfterTlsHandshake(final HttpContext httpContext, final HttpHost endpointHost) {
}
}