View Javadoc

1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  package org.apache.http.impl.conn;
28  
29  import java.io.Closeable;
30  import java.io.IOException;
31  import java.net.InetSocketAddress;
32  import java.util.Map;
33  import java.util.concurrent.ConcurrentHashMap;
34  import java.util.concurrent.ExecutionException;
35  import java.util.concurrent.Future;
36  import java.util.concurrent.TimeUnit;
37  import java.util.concurrent.TimeoutException;
38  import java.util.concurrent.atomic.AtomicBoolean;
39  
40  import org.apache.commons.logging.Log;
41  import org.apache.commons.logging.LogFactory;
42  import org.apache.http.HttpClientConnection;
43  import org.apache.http.HttpHost;
44  import org.apache.http.annotation.ThreadSafe;
45  import org.apache.http.config.ConnectionConfig;
46  import org.apache.http.config.Lookup;
47  import org.apache.http.config.Registry;
48  import org.apache.http.config.RegistryBuilder;
49  import org.apache.http.config.SocketConfig;
50  import org.apache.http.conn.ConnectionPoolTimeoutException;
51  import org.apache.http.conn.ConnectionRequest;
52  import org.apache.http.conn.DnsResolver;
53  import org.apache.http.conn.HttpClientConnectionManager;
54  import org.apache.http.conn.HttpConnectionFactory;
55  import org.apache.http.conn.SchemePortResolver;
56  import org.apache.http.conn.ManagedHttpClientConnection;
57  import org.apache.http.conn.routing.HttpRoute;
58  import org.apache.http.conn.socket.ConnectionSocketFactory;
59  import org.apache.http.conn.socket.PlainConnectionSocketFactory;
60  import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
61  import org.apache.http.pool.ConnFactory;
62  import org.apache.http.pool.ConnPoolControl;
63  import org.apache.http.pool.PoolStats;
64  import org.apache.http.protocol.HttpContext;
65  import org.apache.http.util.Args;
66  import org.apache.http.util.Asserts;
67  
68  /**
69   * <tt>ClientConnectionPoolManager</tt> maintains a pool of
70   * {@link HttpClientConnection}s and is able to service connection requests
71   * from multiple execution threads. Connections are pooled on a per route
72   * basis. A request for a route which already the manager has persistent
73   * connections for available in the pool will be services by leasing
74   * a connection from the pool rather than creating a brand new connection.
75   * <p/>
76   * <tt>ClientConnectionPoolManager</tt> maintains a maximum limit of connection
77   * on a per route basis and in total. Per default this implementation will
78   * create no more than than 2 concurrent connections per given route
79   * and no more 20 connections in total. For many real-world applications
80   * these limits may prove too constraining, especially if they use HTTP
81   * as a transport protocol for their services. Connection limits, however,
82   * can be adjusted using {@link ConnPoolControl} methods.
83   *
84   * @since 4.3
85   */
86  @ThreadSafe
87  public class PoolingHttpClientConnectionManager
88      implements HttpClientConnectionManager, ConnPoolControl<HttpRoute>, Closeable {
89  
90      private final Log log = LogFactory.getLog(getClass());
91  
92      private final ConfigData configData;
93      private final CPool pool;
94      private final HttpClientConnectionOperator connectionOperator;
95      private final AtomicBoolean isShutDown;
96  
97      private static Registry<ConnectionSocketFactory> getDefaultRegistry() {
98          return RegistryBuilder.<ConnectionSocketFactory>create()
99                  .register("http", PlainConnectionSocketFactory.getSocketFactory())
100                 .register("https", SSLConnectionSocketFactory.getSocketFactory())
101                 .build();
102     }
103 
104     public PoolingHttpClientConnectionManager() {
105         this(getDefaultRegistry());
106     }
107 
108     public PoolingHttpClientConnectionManager(final long timeToLive, final TimeUnit tunit) {
109         this(getDefaultRegistry(), null, null ,null, timeToLive, tunit);
110     }
111 
112     public PoolingHttpClientConnectionManager(
113             final Registry<ConnectionSocketFactory> socketFactoryRegistry) {
114         this(socketFactoryRegistry, null, null);
115     }
116 
117     public PoolingHttpClientConnectionManager(
118             final Registry<ConnectionSocketFactory> socketFactoryRegistry,
119             final DnsResolver dnsResolver) {
120         this(socketFactoryRegistry, null, dnsResolver);
121     }
122 
123     public PoolingHttpClientConnectionManager(
124             final Registry<ConnectionSocketFactory> socketFactoryRegistry,
125             final HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connFactory) {
126         this(socketFactoryRegistry, connFactory, null);
127     }
128 
129     public PoolingHttpClientConnectionManager(
130             final HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connFactory) {
131         this(getDefaultRegistry(), connFactory, null);
132     }
133 
134     public PoolingHttpClientConnectionManager(
135             final Registry<ConnectionSocketFactory> socketFactoryRegistry,
136             final HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connFactory,
137             final DnsResolver dnsResolver) {
138         this(socketFactoryRegistry, connFactory, null, dnsResolver, -1, TimeUnit.MILLISECONDS);
139     }
140 
141     public PoolingHttpClientConnectionManager(
142             final Registry<ConnectionSocketFactory> socketFactoryRegistry,
143             final HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connFactory,
144             final SchemePortResolver schemePortResolver,
145             final DnsResolver dnsResolver,
146             final long timeToLive, final TimeUnit tunit) {
147         super();
148         this.configData = new ConfigData();
149         this.pool = new CPool(
150                 new InternalConnectionFactory(this.configData, connFactory), 2, 20, timeToLive, tunit);
151         this.connectionOperator = new HttpClientConnectionOperator(
152                 socketFactoryRegistry, schemePortResolver, dnsResolver);
153         this.isShutDown = new AtomicBoolean(false);
154     }
155 
156     PoolingHttpClientConnectionManager(
157             final CPool pool,
158             final Lookup<ConnectionSocketFactory> socketFactoryRegistry,
159             final SchemePortResolver schemePortResolver,
160             final DnsResolver dnsResolver) {
161         super();
162         this.configData = new ConfigData();
163         this.pool = pool;
164         this.connectionOperator = new HttpClientConnectionOperator(
165                 socketFactoryRegistry, schemePortResolver, dnsResolver);
166         this.isShutDown = new AtomicBoolean(false);
167     }
168 
169     @Override
170     protected void finalize() throws Throwable {
171         try {
172             shutdown();
173         } finally {
174             super.finalize();
175         }
176     }
177 
178     public void close() {
179         shutdown();
180     }
181 
182     private String format(final HttpRoute route, final Object state) {
183         final StringBuilder buf = new StringBuilder();
184         buf.append("[route: ").append(route).append("]");
185         if (state != null) {
186             buf.append("[state: ").append(state).append("]");
187         }
188         return buf.toString();
189     }
190 
191     private String formatStats(final HttpRoute route) {
192         final StringBuilder buf = new StringBuilder();
193         final PoolStats totals = this.pool.getTotalStats();
194         final PoolStats stats = this.pool.getStats(route);
195         buf.append("[total kept alive: ").append(totals.getAvailable()).append("; ");
196         buf.append("route allocated: ").append(stats.getLeased() + stats.getAvailable());
197         buf.append(" of ").append(stats.getMax()).append("; ");
198         buf.append("total allocated: ").append(totals.getLeased() + totals.getAvailable());
199         buf.append(" of ").append(totals.getMax()).append("]");
200         return buf.toString();
201     }
202 
203     private String format(final CPoolEntry entry) {
204         final StringBuilder buf = new StringBuilder();
205         buf.append("[id: ").append(entry.getId()).append("]");
206         buf.append("[route: ").append(entry.getRoute()).append("]");
207         final Object state = entry.getState();
208         if (state != null) {
209             buf.append("[state: ").append(state).append("]");
210         }
211         return buf.toString();
212     }
213 
214     public ConnectionRequest requestConnection(
215             final HttpRoute route,
216             final Object state) {
217         Args.notNull(route, "HTTP route");
218         if (this.log.isDebugEnabled()) {
219             this.log.debug("Connection request: " + format(route, state) + formatStats(route));
220         }
221         final Future<CPoolEntry> future = this.pool.lease(route, state, null);
222         return new ConnectionRequest() {
223 
224             public boolean cancel() {
225                 return future.cancel(true);
226             }
227 
228             public HttpClientConnection get(
229                     final long timeout,
230                     final TimeUnit tunit) throws InterruptedException, ExecutionException, ConnectionPoolTimeoutException {
231                 return leaseConnection(future, timeout, tunit);
232             }
233 
234         };
235 
236     }
237 
238     protected HttpClientConnection leaseConnection(
239             final Future<CPoolEntry> future,
240             final long timeout,
241             final TimeUnit tunit) throws InterruptedException, ExecutionException, ConnectionPoolTimeoutException {
242         final CPoolEntry entry;
243         try {
244             entry = future.get(timeout, tunit);
245             if (entry == null || future.isCancelled()) {
246                 throw new InterruptedException();
247             }
248             Asserts.check(entry.getConnection() != null, "Pool entry with no connection");
249             if (this.log.isDebugEnabled()) {
250                 this.log.debug("Connection leased: " + format(entry) + formatStats(entry.getRoute()));
251             }
252             return CPoolProxy.newProxy(entry);
253         } catch (final TimeoutException ex) {
254             throw new ConnectionPoolTimeoutException("Timeout waiting for connection from pool");
255         }
256     }
257 
258     public void releaseConnection(
259             final HttpClientConnection managedConn,
260             final Object state,
261             final long keepalive, final TimeUnit tunit) {
262         Args.notNull(managedConn, "Managed connection");
263         synchronized (managedConn) {
264             final CPoolEntry entry = CPoolProxy.detach(managedConn);
265             if (entry == null) {
266                 return;
267             }
268             final ManagedHttpClientConnection conn = entry.getConnection();
269             try {
270                 if (conn.isOpen()) {
271                     entry.setState(state);
272                     entry.updateExpiry(keepalive, tunit != null ? tunit : TimeUnit.MILLISECONDS);
273                     if (this.log.isDebugEnabled()) {
274                         final String s;
275                         if (keepalive > 0) {
276                             s = "for " + (double) keepalive / 1000 + " seconds";
277                         } else {
278                             s = "indefinitely";
279                         }
280                         this.log.debug("Connection " + format(entry) + " can be kept alive " + s);
281                     }
282                 }
283             } finally {
284                 this.pool.release(entry, conn.isOpen() && entry.isRouteComplete());
285                 if (this.log.isDebugEnabled()) {
286                     this.log.debug("Connection released: " + format(entry) + formatStats(entry.getRoute()));
287                 }
288             }
289         }
290     }
291 
292     public void connect(
293             final HttpClientConnection managedConn,
294             final HttpRoute route,
295             final int connectTimeout,
296             final HttpContext context) throws IOException {
297         Args.notNull(managedConn, "Managed Connection");
298         Args.notNull(route, "HTTP route");
299         final ManagedHttpClientConnection conn;
300         synchronized (managedConn) {
301             final CPoolEntry entry = CPoolProxy.getPoolEntry(managedConn);
302             conn = entry.getConnection();
303         }
304         final HttpHost host;
305         if (route.getProxyHost() != null) {
306             host = route.getProxyHost();
307         } else {
308             host = route.getTargetHost();
309         }
310         final InetSocketAddress localAddress = route.getLocalSocketAddress();
311         SocketConfig socketConfig = this.configData.getSocketConfig(host);
312         if (socketConfig == null) {
313             socketConfig = this.configData.getDefaultSocketConfig();
314         }
315         if (socketConfig == null) {
316             socketConfig = SocketConfig.DEFAULT;
317         }
318         this.connectionOperator.connect(
319                 conn, host, localAddress, connectTimeout, socketConfig, context);
320     }
321 
322     public void upgrade(
323             final HttpClientConnection managedConn,
324             final HttpRoute route,
325             final HttpContext context) throws IOException {
326         Args.notNull(managedConn, "Managed Connection");
327         Args.notNull(route, "HTTP route");
328         final ManagedHttpClientConnection conn;
329         synchronized (managedConn) {
330             final CPoolEntry entry = CPoolProxy.getPoolEntry(managedConn);
331             conn = entry.getConnection();
332         }
333         this.connectionOperator.upgrade(conn, route.getTargetHost(), context);
334     }
335 
336     public void routeComplete(
337             final HttpClientConnection managedConn,
338             final HttpRoute route,
339             final HttpContext context) throws IOException {
340         Args.notNull(managedConn, "Managed Connection");
341         Args.notNull(route, "HTTP route");
342         synchronized (managedConn) {
343             final CPoolEntry entry = CPoolProxy.getPoolEntry(managedConn);
344             entry.markRouteComplete();
345         }
346     }
347 
348     public void shutdown() {
349         if (this.isShutDown.compareAndSet(false, true)) {
350             this.log.debug("Connection manager is shutting down");
351             try {
352                 this.pool.shutdown();
353             } catch (final IOException ex) {
354                 this.log.debug("I/O exception shutting down connection manager", ex);
355             }
356             this.log.debug("Connection manager shut down");
357         }
358     }
359 
360     public void closeIdleConnections(final long idleTimeout, final TimeUnit tunit) {
361         if (this.log.isDebugEnabled()) {
362             this.log.debug("Closing connections idle longer than " + idleTimeout + " " + tunit);
363         }
364         this.pool.closeIdle(idleTimeout, tunit);
365     }
366 
367     public void closeExpiredConnections() {
368         this.log.debug("Closing expired connections");
369         this.pool.closeExpired();
370     }
371 
372     public int getMaxTotal() {
373         return this.pool.getMaxTotal();
374     }
375 
376     public void setMaxTotal(final int max) {
377         this.pool.setMaxTotal(max);
378     }
379 
380     public int getDefaultMaxPerRoute() {
381         return this.pool.getDefaultMaxPerRoute();
382     }
383 
384     public void setDefaultMaxPerRoute(final int max) {
385         this.pool.setDefaultMaxPerRoute(max);
386     }
387 
388     public int getMaxPerRoute(final HttpRoute route) {
389         return this.pool.getMaxPerRoute(route);
390     }
391 
392     public void setMaxPerRoute(final HttpRoute route, final int max) {
393         this.pool.setMaxPerRoute(route, max);
394     }
395 
396     public PoolStats getTotalStats() {
397         return this.pool.getTotalStats();
398     }
399 
400     public PoolStats getStats(final HttpRoute route) {
401         return this.pool.getStats(route);
402     }
403 
404     public SocketConfig getDefaultSocketConfig() {
405         return this.configData.getDefaultSocketConfig();
406     }
407 
408     public void setDefaultSocketConfig(final SocketConfig defaultSocketConfig) {
409         this.configData.setDefaultSocketConfig(defaultSocketConfig);
410     }
411 
412     public ConnectionConfig getDefaultConnectionConfig() {
413         return this.configData.getDefaultConnectionConfig();
414     }
415 
416     public void setDefaultConnectionConfig(final ConnectionConfig defaultConnectionConfig) {
417         this.configData.setDefaultConnectionConfig(defaultConnectionConfig);
418     }
419 
420     public SocketConfig getSocketConfig(final HttpHost host) {
421         return this.configData.getSocketConfig(host);
422     }
423 
424     public void setSocketConfig(final HttpHost host, final SocketConfig socketConfig) {
425         this.configData.setSocketConfig(host, socketConfig);
426     }
427 
428     public ConnectionConfig getConnectionConfig(final HttpHost host) {
429         return this.configData.getConnectionConfig(host);
430     }
431 
432     public void setConnectionConfig(final HttpHost host, final ConnectionConfig connectionConfig) {
433         this.configData.setConnectionConfig(host, connectionConfig);
434     }
435 
436     static class ConfigData {
437 
438         private final Map<HttpHost, SocketConfig> socketConfigMap;
439         private final Map<HttpHost, ConnectionConfig> connectionConfigMap;
440         private volatile SocketConfig defaultSocketConfig;
441         private volatile ConnectionConfig defaultConnectionConfig;
442 
443         ConfigData() {
444             super();
445             this.socketConfigMap = new ConcurrentHashMap<HttpHost, SocketConfig>();
446             this.connectionConfigMap = new ConcurrentHashMap<HttpHost, ConnectionConfig>();
447         }
448 
449         public SocketConfig getDefaultSocketConfig() {
450             return this.defaultSocketConfig;
451         }
452 
453         public void setDefaultSocketConfig(final SocketConfig defaultSocketConfig) {
454             this.defaultSocketConfig = defaultSocketConfig;
455         }
456 
457         public ConnectionConfig getDefaultConnectionConfig() {
458             return this.defaultConnectionConfig;
459         }
460 
461         public void setDefaultConnectionConfig(final ConnectionConfig defaultConnectionConfig) {
462             this.defaultConnectionConfig = defaultConnectionConfig;
463         }
464 
465         public SocketConfig getSocketConfig(final HttpHost host) {
466             return this.socketConfigMap.get(host);
467         }
468 
469         public void setSocketConfig(final HttpHost host, final SocketConfig socketConfig) {
470             this.socketConfigMap.put(host, socketConfig);
471         }
472 
473         public ConnectionConfig getConnectionConfig(final HttpHost host) {
474             return this.connectionConfigMap.get(host);
475         }
476 
477         public void setConnectionConfig(final HttpHost host, final ConnectionConfig connectionConfig) {
478             this.connectionConfigMap.put(host, connectionConfig);
479         }
480 
481     }
482 
483     static class InternalConnectionFactory implements ConnFactory<HttpRoute, ManagedHttpClientConnection> {
484 
485         private final ConfigData configData;
486         private final HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connFactory;
487 
488         InternalConnectionFactory(
489                 final ConfigData configData,
490                 final HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connFactory) {
491             super();
492             this.configData = configData != null ? configData : new ConfigData();
493             this.connFactory = connFactory != null ? connFactory :
494                 ManagedHttpClientConnectionFactory.INSTANCE;
495         }
496 
497         public ManagedHttpClientConnection create(final HttpRoute route) throws IOException {
498             ConnectionConfig config = null;
499             if (route.getProxyHost() != null) {
500                 config = this.configData.getConnectionConfig(route.getProxyHost());
501             }
502             if (config == null) {
503                 config = this.configData.getConnectionConfig(route.getTargetHost());
504             }
505             if (config == null) {
506                 config = this.configData.getDefaultConnectionConfig();
507             }
508             if (config == null) {
509                 config = ConnectionConfig.DEFAULT;
510             }
511             return this.connFactory.create(route, config);
512         }
513 
514     }
515 
516 }