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.InetSocketAddress;
31 import java.util.Set;
32 import java.util.concurrent.ExecutionException;
33 import java.util.concurrent.Future;
34 import java.util.concurrent.TimeUnit;
35 import java.util.concurrent.TimeoutException;
36 import java.util.concurrent.atomic.AtomicBoolean;
37 import java.util.concurrent.atomic.AtomicReference;
38
39 import org.apache.hc.client5.http.DnsResolver;
40 import org.apache.hc.client5.http.HttpRoute;
41 import org.apache.hc.client5.http.SchemePortResolver;
42 import org.apache.hc.client5.http.config.ConnectionConfig;
43 import org.apache.hc.client5.http.config.TlsConfig;
44 import org.apache.hc.client5.http.impl.ConnPoolSupport;
45 import org.apache.hc.client5.http.impl.ConnectionShutdownException;
46 import org.apache.hc.client5.http.impl.PrefixedIncrementingId;
47 import org.apache.hc.client5.http.nio.AsyncClientConnectionManager;
48 import org.apache.hc.client5.http.nio.AsyncClientConnectionOperator;
49 import org.apache.hc.client5.http.nio.AsyncConnectionEndpoint;
50 import org.apache.hc.client5.http.nio.ManagedAsyncClientConnection;
51 import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
52 import org.apache.hc.core5.annotation.Contract;
53 import org.apache.hc.core5.annotation.Internal;
54 import org.apache.hc.core5.annotation.ThreadingBehavior;
55 import org.apache.hc.core5.concurrent.BasicFuture;
56 import org.apache.hc.core5.concurrent.CallbackContribution;
57 import org.apache.hc.core5.concurrent.ComplexFuture;
58 import org.apache.hc.core5.concurrent.FutureCallback;
59 import org.apache.hc.core5.function.Resolver;
60 import org.apache.hc.core5.http.HttpHost;
61 import org.apache.hc.core5.http.HttpVersion;
62 import org.apache.hc.core5.http.ProtocolVersion;
63 import org.apache.hc.core5.http.URIScheme;
64 import org.apache.hc.core5.http.config.Lookup;
65 import org.apache.hc.core5.http.config.RegistryBuilder;
66 import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
67 import org.apache.hc.core5.http.nio.AsyncPushConsumer;
68 import org.apache.hc.core5.http.nio.HandlerFactory;
69 import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
70 import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
71 import org.apache.hc.core5.http.protocol.HttpContext;
72 import org.apache.hc.core5.http2.HttpVersionPolicy;
73 import org.apache.hc.core5.http2.nio.command.PingCommand;
74 import org.apache.hc.core5.http2.nio.support.BasicPingHandler;
75 import org.apache.hc.core5.http2.ssl.ApplicationProtocol;
76 import org.apache.hc.core5.io.CloseMode;
77 import org.apache.hc.core5.pool.ConnPoolControl;
78 import org.apache.hc.core5.pool.LaxConnPool;
79 import org.apache.hc.core5.pool.ManagedConnPool;
80 import org.apache.hc.core5.pool.PoolConcurrencyPolicy;
81 import org.apache.hc.core5.pool.PoolEntry;
82 import org.apache.hc.core5.pool.PoolReusePolicy;
83 import org.apache.hc.core5.pool.PoolStats;
84 import org.apache.hc.core5.pool.StrictConnPool;
85 import org.apache.hc.core5.reactor.Command;
86 import org.apache.hc.core5.reactor.ConnectionInitiator;
87 import org.apache.hc.core5.reactor.ProtocolIOSession;
88 import org.apache.hc.core5.reactor.ssl.TlsDetails;
89 import org.apache.hc.core5.util.Args;
90 import org.apache.hc.core5.util.Deadline;
91 import org.apache.hc.core5.util.Identifiable;
92 import org.apache.hc.core5.util.TimeValue;
93 import org.apache.hc.core5.util.Timeout;
94 import org.slf4j.Logger;
95 import org.slf4j.LoggerFactory;
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115 @Contract(threading = ThreadingBehavior.SAFE_CONDITIONAL)
116 public class PoolingAsyncClientConnectionManager implements AsyncClientConnectionManager, ConnPoolControl<HttpRoute> {
117
118 private static final Logger LOG = LoggerFactory.getLogger(PoolingAsyncClientConnectionManager.class);
119
120 public static final int DEFAULT_MAX_TOTAL_CONNECTIONS = 25;
121 public static final int DEFAULT_MAX_CONNECTIONS_PER_ROUTE = 5;
122
123 private final ManagedConnPool<HttpRoute, ManagedAsyncClientConnection> pool;
124 private final AsyncClientConnectionOperator connectionOperator;
125 private final AtomicBoolean closed;
126
127 private volatile Resolver<HttpRoute, ConnectionConfig> connectionConfigResolver;
128 private volatile Resolver<HttpHost, TlsConfig> tlsConfigResolver;
129
130 public PoolingAsyncClientConnectionManager() {
131 this(RegistryBuilder.<TlsStrategy>create()
132 .register(URIScheme.HTTPS.getId(), DefaultClientTlsStrategy.getDefault())
133 .build());
134 }
135
136 public PoolingAsyncClientConnectionManager(final Lookup<TlsStrategy> tlsStrategyLookup) {
137 this(tlsStrategyLookup, PoolConcurrencyPolicy.STRICT, TimeValue.NEG_ONE_MILLISECOND);
138 }
139
140 public PoolingAsyncClientConnectionManager(
141 final Lookup<TlsStrategy> tlsStrategyLookup,
142 final PoolConcurrencyPolicy poolConcurrencyPolicy,
143 final TimeValue timeToLive) {
144 this(tlsStrategyLookup, poolConcurrencyPolicy, PoolReusePolicy.LIFO, timeToLive);
145 }
146
147 public PoolingAsyncClientConnectionManager(
148 final Lookup<TlsStrategy> tlsStrategyLookup,
149 final PoolConcurrencyPolicy poolConcurrencyPolicy,
150 final PoolReusePolicy poolReusePolicy,
151 final TimeValue timeToLive) {
152 this(tlsStrategyLookup, poolConcurrencyPolicy, poolReusePolicy, timeToLive, null, null);
153 }
154
155 public PoolingAsyncClientConnectionManager(
156 final Lookup<TlsStrategy> tlsStrategyLookup,
157 final PoolConcurrencyPolicy poolConcurrencyPolicy,
158 final PoolReusePolicy poolReusePolicy,
159 final TimeValue timeToLive,
160 final SchemePortResolver schemePortResolver,
161 final DnsResolver dnsResolver) {
162 this(new DefaultAsyncClientConnectionOperator(tlsStrategyLookup, schemePortResolver, dnsResolver),
163 poolConcurrencyPolicy, poolReusePolicy, timeToLive);
164 }
165
166 @Internal
167 protected PoolingAsyncClientConnectionManager(
168 final AsyncClientConnectionOperator connectionOperator,
169 final PoolConcurrencyPolicy poolConcurrencyPolicy,
170 final PoolReusePolicy poolReusePolicy,
171 final TimeValue timeToLive) {
172 this.connectionOperator = Args.notNull(connectionOperator, "Connection operator");
173 switch (poolConcurrencyPolicy != null ? poolConcurrencyPolicy : PoolConcurrencyPolicy.STRICT) {
174 case STRICT:
175 this.pool = new StrictConnPool<HttpRoute, ManagedAsyncClientConnection>(
176 DEFAULT_MAX_CONNECTIONS_PER_ROUTE,
177 DEFAULT_MAX_TOTAL_CONNECTIONS,
178 timeToLive,
179 poolReusePolicy,
180 null) {
181
182 @Override
183 public void closeExpired() {
184 enumAvailable(e -> closeIfExpired(e));
185 }
186
187 };
188 break;
189 case LAX:
190 this.pool = new LaxConnPool<HttpRoute, ManagedAsyncClientConnection>(
191 DEFAULT_MAX_CONNECTIONS_PER_ROUTE,
192 timeToLive,
193 poolReusePolicy,
194 null) {
195
196 @Override
197 public void closeExpired() {
198 enumAvailable(e -> closeIfExpired(e));
199 }
200
201 };
202 break;
203 default:
204 throw new IllegalArgumentException("Unexpected PoolConcurrencyPolicy value: " + poolConcurrencyPolicy);
205 }
206 this.closed = new AtomicBoolean(false);
207 }
208
209 @Internal
210 protected PoolingAsyncClientConnectionManager(
211 final ManagedConnPool<HttpRoute, ManagedAsyncClientConnection> pool,
212 final AsyncClientConnectionOperator connectionOperator) {
213 this.connectionOperator = Args.notNull(connectionOperator, "Connection operator");
214 this.pool = Args.notNull(pool, "Connection pool");
215 this.closed = new AtomicBoolean(false);
216 }
217
218 @Override
219 public void close() {
220 close(CloseMode.GRACEFUL);
221 }
222
223 @Override
224 public void close(final CloseMode closeMode) {
225 if (this.closed.compareAndSet(false, true)) {
226 if (LOG.isDebugEnabled()) {
227 LOG.debug("Shutdown connection pool {}", closeMode);
228 }
229 this.pool.close(closeMode);
230 LOG.debug("Connection pool shut down");
231 }
232 }
233
234 private InternalConnectionEndpoint cast(final AsyncConnectionEndpoint endpoint) {
235 if (endpoint instanceof InternalConnectionEndpoint) {
236 return (InternalConnectionEndpoint) endpoint;
237 }
238 throw new IllegalStateException("Unexpected endpoint class: " + endpoint.getClass());
239 }
240
241 private ConnectionConfig resolveConnectionConfig(final HttpRoute route) {
242 final Resolver<HttpRoute, ConnectionConfig> resolver = this.connectionConfigResolver;
243 final ConnectionConfig connectionConfig = resolver != null ? resolver.resolve(route) : null;
244 return connectionConfig != null ? connectionConfig : ConnectionConfig.DEFAULT;
245 }
246
247 private TlsConfig resolveTlsConfig(final HttpHost host, final Object attachment) {
248 if (attachment instanceof TlsConfig) {
249 return (TlsConfig) attachment;
250 }
251 final Resolver<HttpHost, TlsConfig> resolver = this.tlsConfigResolver;
252 final TlsConfig tlsConfig = resolver != null ? resolver.resolve(host) : null;
253 return tlsConfig != null ? tlsConfig : TlsConfig.DEFAULT;
254 }
255
256 @Override
257 public Future<AsyncConnectionEndpoint> lease(
258 final String id,
259 final HttpRoute route,
260 final Object state,
261 final Timeout requestTimeout,
262 final FutureCallback<AsyncConnectionEndpoint> callback) {
263 if (LOG.isDebugEnabled()) {
264 LOG.debug("{} endpoint lease request ({}) {}", id, requestTimeout, ConnPoolSupport.formatStats(route, state, pool));
265 }
266 return new Future<AsyncConnectionEndpoint>() {
267
268 final ConnectionConfig connectionConfig = resolveConnectionConfig(route);
269 final BasicFuture<AsyncConnectionEndpoint> resultFuture = new BasicFuture<>(callback);
270
271 final Future<PoolEntry<HttpRoute, ManagedAsyncClientConnection>> leaseFuture = pool.lease(
272 route,
273 state,
274 requestTimeout, new FutureCallback<PoolEntry<HttpRoute, ManagedAsyncClientConnection>>() {
275
276 @Override
277 public void completed(final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry) {
278 if (poolEntry.hasConnection()) {
279 final TimeValue timeToLive = connectionConfig.getTimeToLive();
280 if (TimeValue.isNonNegative(timeToLive)) {
281 final Deadline deadline = Deadline.calculate(poolEntry.getCreated(), timeToLive);
282 if (deadline.isExpired()) {
283 poolEntry.discardConnection(CloseMode.GRACEFUL);
284 }
285 }
286 }
287 if (poolEntry.hasConnection()) {
288 final ManagedAsyncClientConnection connection = poolEntry.getConnection();
289 final TimeValue timeValue = connectionConfig.getValidateAfterInactivity();
290 if (connection.isOpen() && TimeValue.isNonNegative(timeValue)) {
291 final Deadline deadline = Deadline.calculate(poolEntry.getUpdated(), timeValue);
292 if (deadline.isExpired()) {
293 final ProtocolVersion protocolVersion = connection.getProtocolVersion();
294 if (protocolVersion != null && protocolVersion.greaterEquals(HttpVersion.HTTP_2_0)) {
295 connection.submitCommand(new PingCommand(new BasicPingHandler(result -> {
296 if (result == null || !result) {
297 if (LOG.isDebugEnabled()) {
298 LOG.debug("{} connection {} is stale", id, ConnPoolSupport.getId(connection));
299 }
300 poolEntry.discardConnection(CloseMode.GRACEFUL);
301 }
302 leaseCompleted(poolEntry);
303 })), Command.Priority.IMMEDIATE);
304 return;
305 } else {
306 if (LOG.isDebugEnabled()) {
307 LOG.debug("{} connection {} is closed", id, ConnPoolSupport.getId(connection));
308 }
309 poolEntry.discardConnection(CloseMode.IMMEDIATE);
310 }
311 }
312 }
313 }
314 leaseCompleted(poolEntry);
315 }
316
317 void leaseCompleted(final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry) {
318 final ManagedAsyncClientConnection connection = poolEntry.getConnection();
319 if (connection != null) {
320 connection.activate();
321 }
322 if (LOG.isDebugEnabled()) {
323 LOG.debug("{} endpoint leased {}", id, ConnPoolSupport.formatStats(route, state, pool));
324 }
325 final AsyncConnectionEndpoint endpoint = new InternalConnectionEndpoint(poolEntry);
326 if (LOG.isDebugEnabled()) {
327 LOG.debug("{} acquired {}", id, ConnPoolSupport.getId(endpoint));
328 }
329 resultFuture.completed(endpoint);
330 }
331
332 @Override
333 public void failed(final Exception ex) {
334 if (LOG.isDebugEnabled()) {
335 LOG.debug("{} endpoint lease failed", id);
336 }
337 resultFuture.failed(ex);
338 }
339
340 @Override
341 public void cancelled() {
342 if (LOG.isDebugEnabled()) {
343 LOG.debug("{} endpoint lease cancelled", id);
344 }
345 resultFuture.cancel();
346 }
347
348 });
349
350 @Override
351 public AsyncConnectionEndpoint get() throws InterruptedException, ExecutionException {
352 return resultFuture.get();
353 }
354
355 @Override
356 public AsyncConnectionEndpoint get(
357 final long timeout, final TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
358 return resultFuture.get(timeout, unit);
359 }
360
361 @Override
362 public boolean cancel(final boolean mayInterruptIfRunning) {
363 return leaseFuture.cancel(mayInterruptIfRunning);
364 }
365
366 @Override
367 public boolean isDone() {
368 return resultFuture.isDone();
369 }
370
371 @Override
372 public boolean isCancelled() {
373 return resultFuture.isCancelled();
374 }
375
376 };
377 }
378
379 @Override
380 public void release(final AsyncConnectionEndpoint endpoint, final Object state, final TimeValue keepAlive) {
381 Args.notNull(endpoint, "Managed endpoint");
382 Args.notNull(keepAlive, "Keep-alive time");
383 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> entry = cast(endpoint).detach();
384 if (entry == null) {
385 return;
386 }
387 if (LOG.isDebugEnabled()) {
388 LOG.debug("{} releasing endpoint", ConnPoolSupport.getId(endpoint));
389 }
390 final ManagedAsyncClientConnection connection = entry.getConnection();
391 boolean reusable = connection != null && connection.isOpen();
392 try {
393 if (reusable) {
394 entry.updateState(state);
395 entry.updateExpiry(keepAlive);
396 connection.passivate();
397 if (LOG.isDebugEnabled()) {
398 final String s;
399 if (TimeValue.isPositive(keepAlive)) {
400 s = "for " + keepAlive;
401 } else {
402 s = "indefinitely";
403 }
404 LOG.debug("{} connection {} can be kept alive {}", ConnPoolSupport.getId(endpoint), ConnPoolSupport.getId(connection), s);
405 }
406 }
407 } catch (final RuntimeException ex) {
408 reusable = false;
409 throw ex;
410 } finally {
411 pool.release(entry, reusable);
412 if (LOG.isDebugEnabled()) {
413 LOG.debug("{} connection released {}", ConnPoolSupport.getId(endpoint), ConnPoolSupport.formatStats(entry.getRoute(), entry.getState(), pool));
414 }
415 }
416 }
417
418 @Override
419 public Future<AsyncConnectionEndpoint> connect(
420 final AsyncConnectionEndpoint endpoint,
421 final ConnectionInitiator connectionInitiator,
422 final Timeout timeout,
423 final Object attachment,
424 final HttpContext context,
425 final FutureCallback<AsyncConnectionEndpoint> callback) {
426 Args.notNull(endpoint, "Endpoint");
427 Args.notNull(connectionInitiator, "Connection initiator");
428 final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
429 final ComplexFuture<AsyncConnectionEndpoint> resultFuture = new ComplexFuture<>(callback);
430 if (internalEndpoint.isConnected()) {
431 resultFuture.completed(endpoint);
432 return resultFuture;
433 }
434 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry = internalEndpoint.getPoolEntry();
435 final HttpRoute route = poolEntry.getRoute();
436 final HttpHost host;
437 if (route.getProxyHost() != null) {
438 host = route.getProxyHost();
439 } else {
440 host = route.getTargetHost();
441 }
442 final InetSocketAddress localAddress = route.getLocalSocketAddress();
443 final ConnectionConfig connectionConfig = resolveConnectionConfig(route);
444 final TlsConfig tlsConfig = resolveTlsConfig(host, attachment);
445 final Timeout connectTimeout = timeout != null ? timeout : connectionConfig.getConnectTimeout();
446
447 if (LOG.isDebugEnabled()) {
448 LOG.debug("{} connecting endpoint to {} ({})", ConnPoolSupport.getId(endpoint), host, connectTimeout);
449 }
450 final Future<ManagedAsyncClientConnection> connectFuture = connectionOperator.connect(
451 connectionInitiator,
452 host,
453 localAddress,
454 connectTimeout,
455 route.isTunnelled() ? TlsConfig.copy(tlsConfig)
456 .setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_1)
457 .build() : tlsConfig,
458 context,
459 new FutureCallback<ManagedAsyncClientConnection>() {
460
461 @Override
462 public void completed(final ManagedAsyncClientConnection connection) {
463 try {
464 if (LOG.isDebugEnabled()) {
465 LOG.debug("{} connected {}", ConnPoolSupport.getId(endpoint), ConnPoolSupport.getId(connection));
466 }
467 final ProtocolVersion protocolVersion = connection.getProtocolVersion();
468 context.setProtocolVersion(protocolVersion);
469 final Timeout socketTimeout = connectionConfig.getSocketTimeout();
470 if (socketTimeout != null) {
471 connection.setSocketTimeout(socketTimeout);
472 }
473 poolEntry.assignConnection(connection);
474 resultFuture.completed(internalEndpoint);
475 } catch (final RuntimeException ex) {
476 resultFuture.failed(ex);
477 }
478 }
479
480 @Override
481 public void failed(final Exception ex) {
482 resultFuture.failed(ex);
483 }
484
485 @Override
486 public void cancelled() {
487 resultFuture.cancel();
488 }
489 });
490 resultFuture.setDependency(connectFuture);
491 return resultFuture;
492 }
493
494 @Override
495 public void upgrade(
496 final AsyncConnectionEndpoint endpoint,
497 final Object attachment,
498 final HttpContext context,
499 final FutureCallback<AsyncConnectionEndpoint> callback) {
500 Args.notNull(endpoint, "Managed endpoint");
501 final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
502 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry = internalEndpoint.getValidatedPoolEntry();
503 final HttpRoute route = poolEntry.getRoute();
504 final HttpHost host = route.getProxyHost() != null ? route.getProxyHost() : route.getTargetHost();
505 final TlsConfig tlsConfig = resolveTlsConfig(host, attachment);
506 connectionOperator.upgrade(
507 poolEntry.getConnection(),
508 route.getTargetHost(),
509 attachment != null ? attachment : tlsConfig,
510 context,
511 new CallbackContribution<ManagedAsyncClientConnection>(callback) {
512
513 @Override
514 public void completed(final ManagedAsyncClientConnection connection) {
515 if (LOG.isDebugEnabled()) {
516 LOG.debug("{} upgraded {}", ConnPoolSupport.getId(internalEndpoint), ConnPoolSupport.getId(connection));
517 }
518 final TlsDetails tlsDetails = connection.getTlsDetails();
519 if (tlsDetails != null && ApplicationProtocol.HTTP_2.id.equals(tlsDetails.getApplicationProtocol())) {
520 connection.switchProtocol(ApplicationProtocol.HTTP_2.id, new CallbackContribution<ProtocolIOSession>(callback) {
521
522 @Override
523 public void completed(final ProtocolIOSession protocolIOSession) {
524 context.setProtocolVersion(HttpVersion.HTTP_2);
525 if (callback != null) {
526 callback.completed(endpoint);
527 }
528 }
529
530 });
531 } else {
532 if (callback != null) {
533 callback.completed(endpoint);
534 }
535 }
536 }
537 });
538 }
539
540 @Override
541 public void upgrade(final AsyncConnectionEndpoint endpoint, final Object attachment, final HttpContext context) {
542 upgrade(endpoint, attachment, context, null);
543 }
544
545 @Override
546 public Set<HttpRoute> getRoutes() {
547 return pool.getRoutes();
548 }
549
550 @Override
551 public void setMaxTotal(final int max) {
552 pool.setMaxTotal(max);
553 }
554
555 @Override
556 public int getMaxTotal() {
557 return pool.getMaxTotal();
558 }
559
560 @Override
561 public void setDefaultMaxPerRoute(final int max) {
562 pool.setDefaultMaxPerRoute(max);
563 }
564
565 @Override
566 public int getDefaultMaxPerRoute() {
567 return pool.getDefaultMaxPerRoute();
568 }
569
570 @Override
571 public void setMaxPerRoute(final HttpRoute route, final int max) {
572 pool.setMaxPerRoute(route, max);
573 }
574
575 @Override
576 public int getMaxPerRoute(final HttpRoute route) {
577 return pool.getMaxPerRoute(route);
578 }
579
580 @Override
581 public void closeIdle(final TimeValue idletime) {
582 pool.closeIdle(idletime);
583 }
584
585 @Override
586 public void closeExpired() {
587 pool.closeExpired();
588 }
589
590 @Override
591 public PoolStats getTotalStats() {
592 return pool.getTotalStats();
593 }
594
595 @Override
596 public PoolStats getStats(final HttpRoute route) {
597 return pool.getStats(route);
598 }
599
600
601
602
603
604
605 public void setDefaultConnectionConfig(final ConnectionConfig config) {
606 this.connectionConfigResolver = (route) -> config;
607 }
608
609
610
611
612
613
614 public void setConnectionConfigResolver(final Resolver<HttpRoute, ConnectionConfig> connectionConfigResolver) {
615 this.connectionConfigResolver = connectionConfigResolver;
616 }
617
618
619
620
621
622
623 public void setDefaultTlsConfig(final TlsConfig config) {
624 this.tlsConfigResolver = (host) -> config;
625 }
626
627
628
629
630
631
632 public void setTlsConfigResolver(final Resolver<HttpHost, TlsConfig> tlsConfigResolver) {
633 this.tlsConfigResolver = tlsConfigResolver;
634 }
635
636 void closeIfExpired(final PoolEntry<HttpRoute, ManagedAsyncClientConnection > entry) {
637 final long now = System.currentTimeMillis();
638 if (entry.getExpiryDeadline().isBefore(now)) {
639 entry.discardConnection(CloseMode.GRACEFUL);
640 } else {
641 final ConnectionConfig connectionConfig = resolveConnectionConfig(entry.getRoute());
642 final TimeValue timeToLive = connectionConfig.getTimeToLive();
643 if (timeToLive != null && Deadline.calculate(entry.getCreated(), timeToLive).isBefore(now)) {
644 entry.discardConnection(CloseMode.GRACEFUL);
645 }
646 }
647 }
648
649
650
651
652 @Deprecated
653 public TimeValue getValidateAfterInactivity() {
654 return ConnectionConfig.DEFAULT.getValidateAfterInactivity();
655 }
656
657
658
659
660
661
662
663
664
665
666 @Deprecated
667 public void setValidateAfterInactivity(final TimeValue validateAfterInactivity) {
668 setDefaultConnectionConfig(ConnectionConfig.custom()
669 .setValidateAfterInactivity(validateAfterInactivity)
670 .build());
671 }
672
673 private static final PrefixedIncrementingId INCREMENTING_ID = new PrefixedIncrementingId("ep-");
674
675 class InternalConnectionEndpoint extends AsyncConnectionEndpoint implements Identifiable {
676
677 private final AtomicReference<PoolEntry<HttpRoute, ManagedAsyncClientConnection>> poolEntryRef;
678 private final String id;
679
680 InternalConnectionEndpoint(final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry) {
681 this.poolEntryRef = new AtomicReference<>(poolEntry);
682 this.id = INCREMENTING_ID.getNextId();
683 }
684
685 @Override
686 public String getId() {
687 return id;
688 }
689
690 PoolEntry<HttpRoute, ManagedAsyncClientConnection> getPoolEntry() {
691 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry = poolEntryRef.get();
692 if (poolEntry == null) {
693 throw new ConnectionShutdownException();
694 }
695 return poolEntry;
696 }
697
698 PoolEntry<HttpRoute, ManagedAsyncClientConnection> getValidatedPoolEntry() {
699 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry = getPoolEntry();
700 if (poolEntry.getConnection() == null) {
701 throw new ConnectionShutdownException();
702 }
703 return poolEntry;
704 }
705
706 PoolEntry<HttpRoute, ManagedAsyncClientConnection> detach() {
707 return poolEntryRef.getAndSet(null);
708 }
709
710 @Override
711 public void close(final CloseMode closeMode) {
712 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry = poolEntryRef.get();
713 if (poolEntry != null) {
714 if (LOG.isDebugEnabled()) {
715 LOG.debug("{} close {}", id, closeMode);
716 }
717 poolEntry.discardConnection(closeMode);
718 }
719 }
720
721 @Override
722 public boolean isConnected() {
723 final PoolEntry<HttpRoute, ManagedAsyncClientConnection> poolEntry = poolEntryRef.get();
724 if (poolEntry == null) {
725 return false;
726 }
727 final ManagedAsyncClientConnection connection = poolEntry.getConnection();
728 if (connection == null) {
729 return false;
730 }
731 if (!connection.isOpen()) {
732 poolEntry.discardConnection(CloseMode.IMMEDIATE);
733 return false;
734 }
735 return true;
736 }
737
738 @Override
739 public void setSocketTimeout(final Timeout timeout) {
740 getValidatedPoolEntry().getConnection().setSocketTimeout(timeout);
741 }
742
743 @Override
744 public void execute(
745 final String exchangeId,
746 final AsyncClientExchangeHandler exchangeHandler,
747 final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
748 final HttpContext context) {
749 final ManagedAsyncClientConnection connection = getValidatedPoolEntry().getConnection();
750 if (LOG.isDebugEnabled()) {
751 LOG.debug("{} executing exchange {} over {}", id, exchangeId, ConnPoolSupport.getId(connection));
752 }
753 connection.submitCommand(
754 new RequestExecutionCommand(exchangeHandler, pushHandlerFactory, context),
755 Command.Priority.NORMAL);
756 }
757
758 }
759
760
761
762
763
764
765
766
767 boolean isClosed() {
768 return this.closed.get();
769 }
770
771 }