DefaultConnectionPool


1   /**
2    * Redistribution and use of this software and associated documentation
3    * ("Software"), with or without modification, are permitted provided
4    * that the following conditions are met:
5    *
6    * 1. Redistributions of source code must retain copyright
7    *    statements and notices.  Redistributions must also contain a
8    *    copy of this document.
9    *
10   * 2. Redistributions in binary form must reproduce the
11   *    above copyright notice, this list of conditions and the
12   *    following disclaimer in the documentation and/or other
13   *    materials provided with the distribution.
14   *
15   * 3. The name "Exolab" must not be used to endorse or promote
16   *    products derived from this Software without prior written
17   *    permission of Exoffice Technologies.  For written permission,
18   *    please contact info@exolab.org.
19   *
20   * 4. Products derived from this Software may not be called "Exolab"
21   *    nor may "Exolab" appear in their names without prior written
22   *    permission of Exoffice Technologies. Exolab is a registered
23   *    trademark of Exoffice Technologies.
24   *
25   * 5. Due credit should be given to the Exolab Project
26   *    (http://www.exolab.org/).
27   *
28   * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
29   * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
30   * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
31   * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
32   * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
33   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
34   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
35   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
36   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
37   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
38   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
39   * OF THE POSSIBILITY OF SUCH DAMAGE.
40   *
41   * Copyright 2005 (C) Exoffice Technologies Inc. All Rights Reserved.
42   *
43   * $Id: DefaultConnectionPool.java,v 1.5 2005/06/05 13:56:49 tanderson Exp $
44   */
45  package org.exolab.jms.net.connector;
46  
47  import java.security.Principal  ;
48  import java.util.ArrayList  ;
49  import java.util.Collections  ;
50  import java.util.HashMap  ;
51  import java.util.List  ;
52  import java.util.Map  ;
53  
54  import org.apache.commons.logging.Log;
55  import org.apache.commons.logging.LogFactory;
56  import EDU.oswego.cs.dl.util.concurrent.ClockDaemon;
57  
58  import org.exolab.jms.net.uri.URI;
59  import org.exolab.jms.net.util.ThreadFactory;
60  import org.exolab.jms.net.util.Properties;
61  
62  
63  /**
64   * Manages a pool of {@link ManagedConnection} instances, for a particular
65   * {@link ManagedConnectionFactory}.
66   *
67   * @author <a HREF="mailto:tma@netspace.net.au">Tim Anderson</a>
68   * @version $Revision: 1.5 $ $Date: 2005/06/05 13:56:49 $
69   * @see AbstractConnectionManager
70   */
71  class DefaultConnectionPool
72          implements ManagedConnectionAcceptorListener,
73          ManagedConnectionListener, ConnectionPool {
74  
75      /**
76       * The connection factory.
77       */
78      private final ManagedConnectionFactory _factory;
79  
80      /**
81       * Invocation handler to assign to each new connection.
82       */
83      private final InvocationHandler _handler;
84  
85      /**
86       * The connection factory for resolving connections via their URI.
87       */
88      private final ConnectionFactory _resolver;
89  
90      /**
91       * The set of allocated connections.
92       */
93      private List   _connections = Collections.synchronizedList(new ArrayList  ());
94  
95      /**
96       * A map of ManagedConnection -> ManagedConnectionHandle. The handles are
97       * used to reap idle connections.
98       */
99      private Map   _handles = Collections.synchronizedMap(new HashMap  ());
100 
101     /**
102      * The set of connection acceptors.
103      */
104     private List   _acceptors = Collections.synchronizedList(new ArrayList  ());
105 
106     /**
107      * The set of accepted connections.
108      */
109     private List   _accepted = Collections.synchronizedList(new ArrayList  ());
110 
111     /**
112      * The set of all connections, as a map of ManagedConnection -> PoolEntry
113      * instances.
114      */
115     private Map   _entries = Collections.synchronizedMap(new HashMap  ());
116 
117     /**
118      * Clock daemon for periodically running the reaper.
119      */
120     private ClockDaemon _daemon;
121 
122     /**
123      * Interval between reaping dead/idle connections, in milliseconds.
124      * If <code>0</code> indicates not to reap connections.
125      */
126     private final long _reapInterval;
127 
128     /**
129      * The caller event listener.
130      */
131     private volatile CallerListener _listener;
132 
133     /**
134      * Property name prefix for pool configuration items.
135      */
136     private static final String   POOL_PREFIX = "org.exolab.jms.net.pool.";
137 
138     /**
139      * Configuration property to indicate the reap interval.
140      */
141     private static final String   REAP_INTERVAL = "reapInterval";
142 
143     /**
144      * The logger.
145      */
146     private static final Log _log
147             = LogFactory.getLog(DefaultConnectionPool.class);
148 
149 
150     /**
151      * Construct a new <code>DefaultConnectionPool</code>.
152      *
153      * @param factory  the managed connection factory
154      * @param handler  the invocation handler, assigned to each new managed
155      *                 connection
156      * @param resolver the connection factory for resolving connections via
157      *                 their URI
158      * @param properties configuration properties. May be <code>null</code>
159      * @throws ResourceException if any configuration property is invalid
160      */
161     public DefaultConnectionPool(ManagedConnectionFactory factory,
162                                  InvocationHandler handler,
163                                  ConnectionFactory resolver,
164                                  Map   properties) throws ResourceException {
165         if (factory == null) {
166             throw new IllegalArgumentException  ("Argument 'factory' is null");
167         }
168         if (handler == null) {
169             throw new IllegalArgumentException  ("Argument 'handler' is null");
170         }
171         if (resolver == null) {
172             throw new IllegalArgumentException  ("Argument 'resolver' is null");
173         }
174         _factory = factory;
175         _handler = handler;
176         _resolver = resolver;
177 
178         Properties config = new Properties(properties, POOL_PREFIX);
179         int seconds = config.getInt(REAP_INTERVAL, 15);
180         if (seconds < 0) {
181             seconds = 0;
182         }
183         _reapInterval = seconds * 1000;
184 
185     }
186 
187     /**
188      * Creates a new connection.
189      *
190      * @param principal the security principal
191      * @param info      the connection request info
192      * @return a new connection
193      * @throws ResourceException if a connection cannot be established
194      */
195     public ManagedConnection createManagedConnection(Principal   principal,
196                                                      ConnectionRequestInfo info)
197             throws ResourceException {
198         ManagedConnection connection = _factory.createManagedConnection(
199                 principal, info);
200         return add(connection, false);
201     }
202 
203     /**
204      * Creates an acceptor for connections.
205      *
206      * @param authenticator authenticates incoming connections
207      * @param info          the connection request info
208      * @return a new connection acceptor
209      * @throws ResourceException if an acceptor cannot be created
210      */
211     public ManagedConnectionAcceptor createManagedConnectionAcceptor(
212             Authenticator authenticator, ConnectionRequestInfo info)
213             throws ResourceException {
214 
215         ManagedConnectionAcceptor acceptor;
216 
217         acceptor = _factory.createManagedConnectionAcceptor(authenticator,
218                                                             info);
219         _acceptors.add(acceptor);
220         return acceptor;
221     }
222 
223     /**
224      * Returns a matched connection from the set of pooled connections.
225      *
226      * @param principal the security principal
227      * @param info      the connection request info
228      * @return the first acceptable match, or <code>null</code> if none is
229      *         found
230      * @throws ResourceException for any error
231      */
232     public ManagedConnection matchManagedConnections(Principal   principal,
233                                                      ConnectionRequestInfo info)
234             throws ResourceException {
235 
236         ManagedConnection result = null;
237         result = _factory.matchManagedConnections(_connections, principal,
238                                                   info);
239         if (result != null) {
240             // return the handle corresponding to the connection
241             result = (ManagedConnection) _handles.get(result);
242         } else {
243             result = _factory.matchManagedConnections(_accepted, principal,
244                                                       info);
245         }
246         return result;
247     }
248 
249     /**
250      * Returns a matched acceptor from the set of pooled connections.
251      *
252      * @param info the connection request info
253      * @return the first acceptable match, or <code>null</code> if none is
254      *         found
255      * @throws ResourceException for any error
256      */
257     public ManagedConnectionAcceptor matchManagedConnectionAcceptors(
258             ConnectionRequestInfo info) throws ResourceException {
259 
260         return _factory.matchManagedConnectionAcceptors(_acceptors, info);
261     }
262 
263     /**
264      * Returns a listener for handling accepted connections.
265      *
266      * @return a listener for handling accepted connections
267      */
268     public ManagedConnectionAcceptorListener
269             getManagedConnectionAcceptorListener() {
270             return this;
271     }
272 
273     /**
274      * Invoked when a new connection is accepted.
275      *
276      * @param acceptor   the acceptor which created the connection
277      * @param connection the accepted connection
278      */
279     public void accepted(ManagedConnectionAcceptor acceptor,
280                          ManagedConnection connection) {
281         try {
282             add(connection, true);
283         } catch (ResourceException exception) {
284             _log.debug("Failed to accept connection", exception);
285         }
286     }
287 
288     /**
289      * Notifies closure of a connection. The <code>ManagedConnection</code>
290      * instance invokes this to notify its registered listeners when the peer
291      * closes the connection.
292      *
293      * @param source the managed connection that is the source of the event
294      */
295     public void closed(ManagedConnection source) {
296         if (_log.isDebugEnabled()) {
297             _log.debug("Connection " + source + " closed by peer, destroying");
298         }
299         remove(source);
300     }
301 
302     /**
303      * Notifies a connection related error. The <code>ManagedConnection</code>
304      * instance invokes this to notify of the occurrence of a physical
305      * connection-related error.
306      *
307      * @param source    the managed connection that is the source of the event
308      * @param throwable the error
309      */
310     public void error(ManagedConnection source, Throwable   throwable) {
311         if (_log.isDebugEnabled()) {
312             _log.debug("Error on connection " + source + ", destroying",
313                        throwable);
314         }
315         remove(source);
316     }
317 
318     /**
319      * Closes this connection pool, cleaning up any allocated resources.
320      *
321      * @throws ResourceException for any error
322      */
323     public void close() throws ResourceException {
324         ManagedConnectionAcceptor[] acceptors =
325                 (ManagedConnectionAcceptor[]) _acceptors.toArray(
326                         new ManagedConnectionAcceptor[0]);
327         _acceptors.clear();
328 
329         for (int i = 0; i < acceptors.length; ++i) {
330             acceptors[i].close();
331         }
332 
333         ManagedConnection[] connections =
334                 (ManagedConnection[]) _entries.keySet().toArray(
335                         new ManagedConnection[0]);
336         for (int i = 0; i < connections.length; ++i) {
337             connections[i].destroy();
338         }
339         _entries.clear();
340 
341         _accepted.clear();
342         _connections.clear();
343 
344         stopReaper();
345     }
346 
347     /**
348      * Invoked when an acceptor receives an error.
349      *
350      * @param acceptor  the acceptor which received the error
351      * @param throwable the error
352      */
353     public void error(ManagedConnectionAcceptor acceptor,
354                       Throwable   throwable) {
355         _acceptors.remove(acceptor);
356 
357         String   uri = "<unknown>";
358         try {
359             uri = acceptor.getURI().toString();
360         } catch (ResourceException ignore) {
361         }
362         _log.error("Failed to accept connections on URI=" + uri,
363                    throwable);
364 
365         try {
366             acceptor.close();
367         } catch (ResourceException exception) {
368             if (_log.isDebugEnabled()) {
369                 _log.debug("Failed to close acceptor, URI=" + uri, exception);
370             }
371         }
372     }
373 
374     /**
375      * Sets the listener for caller events.
376      *
377      * @param listener the listener
378      */
379     public void setCallerListener(CallerListener listener) {
380         _listener = listener;
381     }
382 
383     /**
384      * Adds a connection to the pool. If the connection was created, a {@link
385      * ManagedConnectionHandle} will be returned, wrapping the supplied
386      * connection.
387      *
388      * @param connection the connection to add
389      * @param accepted  if <code>true</code> the connection was accepted via an
390      *                  {@link ManagedConnectionAcceptor}, otherwise it was
391      *                  created via
392      *                  {@link ManagedConnectionFactory#createManagedConnection}
393      * @return the (possibly wrapped) connection
394      * @throws ResourceException if the connection cannot be added
395      */
396     protected ManagedConnection add(ManagedConnection connection,
397                                     boolean accepted) throws ResourceException {
398         ManagedConnection result;
399 
400         PoolEntry entry = new PoolEntry(connection, accepted);
401         _entries.put(connection, entry);
402         if (accepted) {
403             _accepted.add(connection);
404             result = connection;
405         } else {
406             _connections.add(connection);
407             ManagedConnection handle = new ManagedConnectionHandle(
408                     connection, _resolver);
409             _handles.put(connection, handle);
410             result = handle;
411         }
412         ContextInvocationHandler handler = new ContextInvocationHandler(
413                 _handler, _resolver);
414         try {
415             connection.setInvocationHandler(handler);
416             connection.setConnectionEventListener(this);
417             handler.setConnection(connection.getConnection());
418         } catch (ResourceException exception) {
419             try {
420                 _log.debug("Failed to initialise connection, destroying",
421                            exception);
422                 connection.destroy();
423             } catch (ResourceException nested) {
424                 _log.debug("Failed to destroy connection", nested);
425             } finally {
426                 _entries.remove(connection);
427                 if (accepted) {
428                     _accepted.remove(connection);
429                 } else {
430                     _connections.remove(connection);
431                     _handles.remove(connection);
432                 }
433             }
434             // propagate the exception
435             throw exception;
436         }
437 
438         // mark the connection as initialised and therefore available for
439         // reaping
440         entry.setInitialised();
441 
442         startReaper();
443 
444         return result;
445     }
446 
447     /**
448      * Remove a connection from the pool.
449      *
450      * @param connection the connection to remove
451      */
452     protected void remove(ManagedConnection connection) {
453         PoolEntry entry = (PoolEntry) _entries.remove(connection);
454         if (entry != null) {
455             if (entry.getAccepted()) {
456                 _accepted.remove(connection);
457             } else {
458                 _connections.remove(connection);
459                 _handles.remove(connection);
460             }
461             URI remoteURI = null;
462             URI localURI = null;
463             try {
464                 remoteURI = connection.getRemoteURI();
465                 localURI = connection.getLocalURI();
466             } catch (ResourceException exception) {
467                 _log.debug("Failed to get connection URIs", exception);
468             }
469 
470             try {
471                 connection.destroy();
472             } catch (ResourceException exception) {
473                 _log.debug("Failed to destroy connection", exception);
474             }
475             if (remoteURI != null && localURI != null) {
476                 notifyDisconnection(remoteURI, localURI);
477             }
478         } else {
479             _log.debug("ManagedConnection not found");
480         }
481         if (_entries.isEmpty()) {
482             stopReaper();
483         }
484     }
485 
486     /**
487      * Notify of a disconnection.
488      *
489      * @param remoteURI the remote address that the client is calling from
490      * @param localURI  the local address that the client is calling to
491      */
492     private void notifyDisconnection(URI remoteURI, URI localURI) {
493         CallerListener listener = _listener;
494         if (listener != null) {
495             listener.disconnected(new CallerImpl(remoteURI, localURI));
496         }
497     }
498 
499     /**
500      * Starts the reaper for dead/idle connections, if needed.
501      */
502     private synchronized void startReaper() {
503         if (_daemon == null && _reapInterval > 0) {
504             _daemon = new ClockDaemon();
505             ThreadFactory creator =
506                     new ThreadFactory(null, "ManagedConnectionReaper", false);
507             _daemon.setThreadFactory(creator);
508             _daemon.executePeriodically(_reapInterval, new Reaper(), false);
509         }
510     }
511 
512     /**
513      * Stops the reaper for dead/idle connections, if needed.
514      */
515     private synchronized void stopReaper() {
516         if (_daemon != null) {
517             _daemon.shutDown();
518             _daemon = null;
519         }
520     }
521 
522     /**
523      * Helper class for reaping idle and dead connections.
524      */
525     private class Reaper implements Runnable   {
526 
527         /**
528          * The time that the reaper last ran, in milliseconds.
529          */
530         private long _lastReapTimestamp = 0;
531 
532         /**
533          * Construct a new <code>Reaper</code>.
534          */
535         public Reaper() {
536             _lastReapTimestamp = System.currentTimeMillis();
537         }
538 
539         /**
540          * Run the reaper.
541          */
542         public void run() {
543             try {
544                 reapIdleConnections();
545                 if (!done()) {
546                     reapDeadConnections();
547                 }
548             } catch (Throwable   exception) {
549                 _log.error(exception, exception);
550             }
551             _lastReapTimestamp = System.currentTimeMillis();
552         }
553 
554         /**
555          * Reap idle connections.
556          */
557         private void reapIdleConnections() {
558             Map.Entry  [] entries = (Map.Entry  []) _handles.entrySet().toArray(
559                     new Map.Entry  [0]);
560             for (int i = 0; i < entries.length && !done(); ++i) {
561                 Map.Entry   entry = entries[i];
562                 ManagedConnection connection =
563                         (ManagedConnection) entry.getKey();
564                 PoolEntry pooled = (PoolEntry) _entries.get(connection);
565                 if (pooled != null && pooled.isInitialised()) {
566                     ManagedConnectionHandle handle =
567                             (ManagedConnectionHandle) entry.getValue();
568                     if (handle.canDestroy() && idle(handle)) {
569                         if (_log.isDebugEnabled()) {
570                             try {
571                                 _log.debug("Reaping idle connection, URI="
572                                            + connection.getRemoteURI()
573                                            + ", local URI="
574                                            + connection.getLocalURI());
575                             } catch (ResourceException ignore) {
576                                 // do nothing
577                             }
578                         }
579                         remove(connection);
580                     }
581                 }
582             }
583         }
584 
585         /**
586          * Determines if a connection is idle.
587          *
588          * @param handle the handle to the underlying managed connection
589          * @return <code>true</code> if the underlying connection is idle,
590          * otherwise <code>false</code>
591          */
592         private boolean idle(ManagedConnectionHandle handle) {
593             boolean result = false;
594             long timestamp = handle.getLastUsedTimestamp();
595             if (timestamp == 0) {
596                 // connection not used yet. Update it, but don't reap it on this
597                 // invocation.
598                 handle.updateLastUsedTimestamp();
599             } else if (timestamp < _lastReapTimestamp) {
600                 result = true;
601             }
602             return result;
603         }
604 
605         /**
606          * Reap dead connections.
607          */
608         private void reapDeadConnections() {
609             PoolEntry[] entries = (PoolEntry[]) _entries.values().toArray(
610                     new PoolEntry[0]);
611             for (int i = 0; i < entries.length && !done(); ++i) {
612                 PoolEntry entry = entries[i];
613                 if (entry.isInitialised()) {
614                     ManagedConnection connection = entry.getManagedConnection();
615                     if (!connection.isAlive()) {
616                         if (_log.isDebugEnabled()) {
617                             try {
618                                 _log.debug("Reaping dead connection, URI="
619                                            + connection.getRemoteURI()
620                                            + ", local URI="
621                                            + connection.getLocalURI());
622                             } catch (ResourceException ignore) {
623                                 // do nothing
624                             }
625                         }
626                         remove(connection);
627                     }
628                 }
629             }
630         }
631 
632         /**
633          * Determines if the reaper should terminate, by chaecking the interrupt
634          * status of the current thread.
635          *
636          * @return <code>true</code> if the reaper should terminate
637          */
638         private boolean done() {
639             return Thread.currentThread().isInterrupted();
640         }
641     }
642 
643 }
644
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags