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