1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28 package org.apache.hc.client5.http.impl.nio;
29
30 import java.net.InetAddress;
31 import java.net.InetSocketAddress;
32 import java.net.SocketAddress;
33 import java.util.concurrent.Future;
34
35 import org.apache.hc.client5.http.DnsResolver;
36 import org.apache.hc.client5.http.SchemePortResolver;
37 import org.apache.hc.client5.http.UnsupportedSchemeException;
38 import org.apache.hc.client5.http.config.TlsConfig;
39 import org.apache.hc.client5.http.impl.ConnPoolSupport;
40 import org.apache.hc.client5.http.impl.DefaultSchemePortResolver;
41 import org.apache.hc.client5.http.nio.AsyncClientConnectionOperator;
42 import org.apache.hc.client5.http.nio.ManagedAsyncClientConnection;
43 import org.apache.hc.client5.http.routing.RoutingSupport;
44 import org.apache.hc.core5.concurrent.CallbackContribution;
45 import org.apache.hc.core5.concurrent.ComplexFuture;
46 import org.apache.hc.core5.concurrent.FutureCallback;
47 import org.apache.hc.core5.concurrent.FutureContribution;
48 import org.apache.hc.core5.http.HttpHost;
49 import org.apache.hc.core5.http.URIScheme;
50 import org.apache.hc.core5.http.config.Lookup;
51 import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
52 import org.apache.hc.core5.http.protocol.HttpContext;
53 import org.apache.hc.core5.net.NamedEndpoint;
54 import org.apache.hc.core5.reactor.ConnectionInitiator;
55 import org.apache.hc.core5.reactor.IOSession;
56 import org.apache.hc.core5.reactor.ssl.TransportSecurityLayer;
57 import org.apache.hc.core5.util.Args;
58 import org.apache.hc.core5.util.Timeout;
59 import org.slf4j.Logger;
60 import org.slf4j.LoggerFactory;
61
62 final class DefaultAsyncClientConnectionOperator implements AsyncClientConnectionOperator {
63
64 private static final Logger LOG = LoggerFactory.getLogger(DefaultAsyncClientConnectionOperator.class);
65
66 private final SchemePortResolver schemePortResolver;
67 private final MultihomeIOSessionRequester sessionRequester;
68 private final Lookup<TlsStrategy> tlsStrategyLookup;
69
70 DefaultAsyncClientConnectionOperator(
71 final Lookup<TlsStrategy> tlsStrategyLookup,
72 final SchemePortResolver schemePortResolver,
73 final DnsResolver dnsResolver) {
74 this.tlsStrategyLookup = Args.notNull(tlsStrategyLookup, "TLS strategy lookup");
75 this.schemePortResolver = schemePortResolver != null ? schemePortResolver : DefaultSchemePortResolver.INSTANCE;
76 this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver);
77 }
78
79 @Override
80 public Future<ManagedAsyncClientConnection> connect(
81 final ConnectionInitiator connectionInitiator,
82 final HttpHost host,
83 final SocketAddress localAddress,
84 final Timeout connectTimeout,
85 final Object attachment,
86 final FutureCallback<ManagedAsyncClientConnection> callback) {
87 return connect(connectionInitiator, host, null, localAddress, connectTimeout,
88 attachment, null, callback);
89 }
90
91 @Override
92 public Future<ManagedAsyncClientConnection> connect(
93 final ConnectionInitiator connectionInitiator,
94 final HttpHost endpointHost,
95 final NamedEndpoint endpointName,
96 final SocketAddress localAddress,
97 final Timeout connectTimeout,
98 final Object attachment,
99 final HttpContext context,
100 final FutureCallback<ManagedAsyncClientConnection> callback) {
101 Args.notNull(connectionInitiator, "Connection initiator");
102 Args.notNull(endpointHost, "Host");
103 final ComplexFuture<ManagedAsyncClientConnection> future = new ComplexFuture<>(callback);
104 final HttpHost remoteEndpoint = RoutingSupport.normalize(endpointHost, schemePortResolver);
105 final InetAddress remoteAddress = endpointHost.getAddress();
106 final TlsConfig tlsConfig = attachment instanceof TlsConfig ? (TlsConfig) attachment : TlsConfig.DEFAULT;
107
108 if (LOG.isDebugEnabled()) {
109 LOG.debug("{} connecting {}->{} ({})", endpointHost, localAddress, remoteAddress, connectTimeout);
110 }
111
112 final Future<IOSession> sessionFuture = sessionRequester.connect(
113 connectionInitiator,
114 remoteEndpoint,
115 remoteAddress != null ? new InetSocketAddress(remoteAddress, remoteEndpoint.getPort()) : null,
116 localAddress,
117 connectTimeout,
118 tlsConfig.getHttpVersionPolicy(),
119 new FutureCallback<IOSession>() {
120
121 @Override
122 public void completed(final IOSession session) {
123 final DefaultManagedAsyncClientConnection connection = new DefaultManagedAsyncClientConnection(session);
124 if (LOG.isDebugEnabled()) {
125 LOG.debug("{} {} connected {}->{}", ConnPoolSupport.getId(connection), endpointHost,
126 connection.getLocalAddress(), connection.getRemoteAddress());
127 }
128 final TlsStrategy tlsStrategy = tlsStrategyLookup != null ? tlsStrategyLookup.lookup(endpointHost.getSchemeName()) : null;
129 if (tlsStrategy != null) {
130 try {
131 final Timeout socketTimeout = connection.getSocketTimeout();
132 final Timeout handshakeTimeout = tlsConfig.getHandshakeTimeout();
133 final NamedEndpoint tlsName = endpointName != null ? endpointName : endpointHost;
134 if (LOG.isDebugEnabled()) {
135 LOG.debug("{} {} upgrading to TLS", ConnPoolSupport.getId(connection), tlsName);
136 }
137 tlsStrategy.upgrade(
138 connection,
139 tlsName,
140 attachment,
141 handshakeTimeout != null ? handshakeTimeout : connectTimeout,
142 new FutureContribution<TransportSecurityLayer>(future) {
143
144 @Override
145 public void completed(final TransportSecurityLayer transportSecurityLayer) {
146 connection.setSocketTimeout(socketTimeout);
147 future.completed(connection);
148 }
149
150 });
151 } catch (final Exception ex) {
152 future.failed(ex);
153 }
154 } else {
155 future.completed(connection);
156 }
157 }
158
159 @Override
160 public void failed(final Exception ex) {
161 future.failed(ex);
162 }
163
164 @Override
165 public void cancelled() {
166 future.cancel();
167 }
168
169 });
170 future.setDependency(sessionFuture);
171 return future;
172 }
173
174 @Override
175 public void upgrade(
176 final ManagedAsyncClientConnection connection,
177 final HttpHost host,
178 final Object attachment) {
179 upgrade(connection, host, null, attachment, null, null);
180 }
181
182 @Override
183 public void upgrade(
184 final ManagedAsyncClientConnection connection,
185 final HttpHost endpointHost,
186 final NamedEndpoint endpointName,
187 final Object attachment,
188 final HttpContext context,
189 final FutureCallback<ManagedAsyncClientConnection> callback) {
190 final String newProtocol = URIScheme.HTTP.same(endpointHost.getSchemeName()) ? URIScheme.HTTPS.id : endpointHost.getSchemeName();
191 final TlsStrategy tlsStrategy = tlsStrategyLookup != null ? tlsStrategyLookup.lookup(newProtocol) : null;
192 if (tlsStrategy != null) {
193 final NamedEndpoint tlsName = endpointName != null ? endpointName : endpointHost;
194 if (LOG.isDebugEnabled()) {
195 LOG.debug("{} {} upgrading to TLS", ConnPoolSupport.getId(connection), tlsName);
196 }
197 tlsStrategy.upgrade(
198 connection,
199 tlsName,
200 attachment,
201 null,
202 new CallbackContribution<TransportSecurityLayer>(callback) {
203
204 @Override
205 public void completed(final TransportSecurityLayer transportSecurityLayer) {
206 if (callback != null) {
207 callback.completed(connection);
208 }
209 }
210
211 });
212 } else {
213 callback.failed(new UnsupportedSchemeException(newProtocol + " protocol is not supported"));
214 }
215 }
216
217 }