001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.jms.pool;
018
019import java.util.Properties;
020import java.util.concurrent.atomic.AtomicBoolean;
021import java.util.concurrent.atomic.AtomicReference;
022
023import javax.jms.Connection;
024import javax.jms.ConnectionFactory;
025import javax.jms.JMSException;
026import javax.jms.QueueConnection;
027import javax.jms.QueueConnectionFactory;
028import javax.jms.TopicConnection;
029import javax.jms.TopicConnectionFactory;
030
031import org.apache.commons.pool.KeyedPoolableObjectFactory;
032import org.apache.commons.pool.impl.GenericKeyedObjectPool;
033import org.slf4j.Logger;
034import org.slf4j.LoggerFactory;
035
036/**
037 * A JMS provider which pools Connection, Session and MessageProducer instances
038 * so it can be used with tools like <a href="http://camel.apache.org/activemq.html">Camel</a> and Spring's
039 * <a href="http://activemq.apache.org/spring-support.html">JmsTemplate and MessagListenerContainer</a>.
040 * Connections, sessions and producers are returned to a pool after use so that they can be reused later
041 * without having to undergo the cost of creating them again.
042 *
043 * b>NOTE:</b> while this implementation does allow the creation of a collection of active consumers,
044 * it does not 'pool' consumers. Pooling makes sense for connections, sessions and producers, which
045 * are expensive to create and can remain idle a minimal cost. Consumers, on the other hand, are usually
046 * just created at startup and left active, handling incoming messages as they come. When a consumer is
047 * complete, it is best to close it rather than return it to a pool for later reuse: this is because,
048 * even if a consumer is idle, ActiveMQ will keep delivering messages to the consumer's prefetch buffer,
049 * where they'll get held until the consumer is active again.
050 *
051 * If you are creating a collection of consumers (for example, for multi-threaded message consumption), you
052 * might want to consider using a lower prefetch value for each consumer (e.g. 10 or 20), to ensure that
053 * all messages don't end up going to just one of the consumers. See this FAQ entry for more detail:
054 * http://activemq.apache.org/i-do-not-receive-messages-in-my-second-consumer.html
055 *
056 * Optionally, one may configure the pool to examine and possibly evict objects as they sit idle in the
057 * pool. This is performed by an "idle object eviction" thread, which runs asynchronously. Caution should
058 * be used when configuring this optional feature. Eviction runs contend with client threads for access
059 * to objects in the pool, so if they run too frequently performance issues may result. The idle object
060 * eviction thread may be configured using the {@link org.apache.activemq.jms.pool.PooledConnectionFactory#setTimeBetweenExpirationCheckMillis} method.  By
061 * default the value is -1 which means no eviction thread will be run.  Set to a non-negative value to
062 * configure the idle eviction thread to run.
063 *
064 */
065public class PooledConnectionFactory implements ConnectionFactory, QueueConnectionFactory, TopicConnectionFactory {
066    private static final transient Logger LOG = LoggerFactory.getLogger(PooledConnectionFactory.class);
067
068    protected final AtomicBoolean stopped = new AtomicBoolean(false);
069    private GenericKeyedObjectPool<ConnectionKey, ConnectionPool> connectionsPool;
070
071    protected Object connectionFactory;
072
073    private int maximumActiveSessionPerConnection = 500;
074    private int idleTimeout = 30 * 1000;
075    private boolean blockIfSessionPoolIsFull = true;
076    private long blockIfSessionPoolIsFullTimeout = -1L;
077    private long expiryTimeout = 0l;
078    private boolean createConnectionOnStartup = true;
079    private boolean useAnonymousProducers = true;
080    private boolean reconnectOnException = true;
081
082    // Temporary value used to always fetch the result of makeObject.
083    private final AtomicReference<ConnectionPool> mostRecentlyCreated = new AtomicReference<ConnectionPool>(null);
084
085    public void initConnectionsPool() {
086        if (this.connectionsPool == null) {
087            this.connectionsPool = new GenericKeyedObjectPool<ConnectionKey, ConnectionPool>(
088                new KeyedPoolableObjectFactory<ConnectionKey, ConnectionPool>() {
089
090                    @Override
091                    public void activateObject(ConnectionKey key, ConnectionPool connection) throws Exception {
092                    }
093
094                    @Override
095                    public void destroyObject(ConnectionKey key, ConnectionPool connection) throws Exception {
096                        try {
097                            if (LOG.isTraceEnabled()) {
098                                LOG.trace("Destroying connection: {}", connection);
099                            }
100                            connection.close();
101                        } catch (Exception e) {
102                            LOG.warn("Close connection failed for connection: " + connection + ". This exception will be ignored.",e);
103                        }
104                    }
105
106                    @Override
107                    public ConnectionPool makeObject(ConnectionKey key) throws Exception {
108                        Connection delegate = createConnection(key);
109
110                        ConnectionPool connection = createConnectionPool(delegate);
111                        connection.setIdleTimeout(getIdleTimeout());
112                        connection.setExpiryTimeout(getExpiryTimeout());
113                        connection.setMaximumActiveSessionPerConnection(getMaximumActiveSessionPerConnection());
114                        connection.setBlockIfSessionPoolIsFull(isBlockIfSessionPoolIsFull());
115                        if (isBlockIfSessionPoolIsFull() && getBlockIfSessionPoolIsFullTimeout() > 0) {
116                            connection.setBlockIfSessionPoolIsFullTimeout(getBlockIfSessionPoolIsFullTimeout());
117                        }
118                        connection.setUseAnonymousProducers(isUseAnonymousProducers());
119                        connection.setReconnectOnException(isReconnectOnException());
120
121                        if (LOG.isTraceEnabled()) {
122                            LOG.trace("Created new connection: {}", connection);
123                        }
124
125                        PooledConnectionFactory.this.mostRecentlyCreated.set(connection);
126
127                        return connection;
128                    }
129
130                    @Override
131                    public void passivateObject(ConnectionKey key, ConnectionPool connection) throws Exception {
132                    }
133
134                    @Override
135                    public boolean validateObject(ConnectionKey key, ConnectionPool connection) {
136                        if (connection != null && connection.expiredCheck()) {
137                            if (LOG.isTraceEnabled()) {
138                                LOG.trace("Connection has expired: {} and will be destroyed", connection);
139                            }
140
141                            return false;
142                        }
143
144                        return true;
145                    }
146                });
147
148            // Set max idle (not max active) since our connections always idle in the pool.
149            this.connectionsPool.setMaxIdle(1);
150            this.connectionsPool.setLifo(false);
151
152            // We always want our validate method to control when idle objects are evicted.
153            this.connectionsPool.setTestOnBorrow(true);
154            this.connectionsPool.setTestWhileIdle(true);
155        }
156    }
157
158    /**
159     * @return the currently configured ConnectionFactory used to create the pooled Connections.
160     */
161    public Object getConnectionFactory() {
162        return connectionFactory;
163    }
164
165    /**
166     * Sets the ConnectionFactory used to create new pooled Connections.
167     * <p/>
168     * Updates to this value do not affect Connections that were previously created and placed
169     * into the pool.  In order to allocate new Connections based off this new ConnectionFactory
170     * it is first necessary to {@link #clear} the pooled Connections.
171     *
172     * @param toUse
173     *      The factory to use to create pooled Connections.
174     */
175    public void setConnectionFactory(final Object toUse) {
176        if (toUse instanceof ConnectionFactory) {
177            this.connectionFactory = toUse;
178        } else {
179            throw new IllegalArgumentException("connectionFactory should implement javax.jmx.ConnectionFactory");
180        }
181    }
182
183    @Override
184    public QueueConnection createQueueConnection() throws JMSException {
185        return (QueueConnection) createConnection();
186    }
187
188    @Override
189    public QueueConnection createQueueConnection(String userName, String password) throws JMSException {
190        return (QueueConnection) createConnection(userName, password);
191    }
192
193    @Override
194    public TopicConnection createTopicConnection() throws JMSException {
195        return (TopicConnection) createConnection();
196    }
197
198    @Override
199    public TopicConnection createTopicConnection(String userName, String password) throws JMSException {
200        return (TopicConnection) createConnection(userName, password);
201    }
202
203    @Override
204    public Connection createConnection() throws JMSException {
205        return createConnection(null, null);
206    }
207
208    @Override
209    public synchronized Connection createConnection(String userName, String password) throws JMSException {
210        if (stopped.get()) {
211            LOG.debug("PooledConnectionFactory is stopped, skip create new connection.");
212            return null;
213        }
214
215        ConnectionPool connection = null;
216        ConnectionKey key = new ConnectionKey(userName, password);
217
218        // This will either return an existing non-expired ConnectionPool or it
219        // will create a new one to meet the demand.
220        if (getConnectionsPool().getNumIdle(key) < getMaxConnections()) {
221            try {
222                connectionsPool.addObject(key);
223                connection = mostRecentlyCreated.getAndSet(null);
224                connection.incrementReferenceCount();
225            } catch (Exception e) {
226                throw createJmsException("Error while attempting to add new Connection to the pool", e);
227            }
228        } else {
229            try {
230                // We can race against other threads returning the connection when there is an
231                // expiration or idle timeout.  We keep pulling out ConnectionPool instances until
232                // we win and get a non-closed instance and then increment the reference count
233                // under lock to prevent another thread from triggering an expiration check and
234                // pulling the rug out from under us.
235                while (connection == null) {
236                    connection = connectionsPool.borrowObject(key);
237                    synchronized (connection) {
238                        if (connection.getConnection() != null) {
239                            connection.incrementReferenceCount();
240                            break;
241                        }
242
243                        // Return the bad one to the pool and let if get destroyed as normal.
244                        connectionsPool.returnObject(key, connection);
245                        connection = null;
246                    }
247                }
248            } catch (Exception e) {
249                throw createJmsException("Error while attempting to retrieve a connection from the pool", e);
250            }
251
252            try {
253                connectionsPool.returnObject(key, connection);
254            } catch (Exception e) {
255                throw createJmsException("Error when returning connection to the pool", e);
256            }
257        }
258
259        return newPooledConnection(connection);
260    }
261
262    protected Connection newPooledConnection(ConnectionPool connection) {
263        return new PooledConnection(connection);
264    }
265
266    private JMSException createJmsException(String msg, Exception cause) {
267        JMSException exception = new JMSException(msg);
268        exception.setLinkedException(cause);
269        exception.initCause(cause);
270        return exception;
271    }
272
273    protected Connection createConnection(ConnectionKey key) throws JMSException {
274        if (connectionFactory instanceof ConnectionFactory) {
275            if (key.getUserName() == null && key.getPassword() == null) {
276                return ((ConnectionFactory) connectionFactory).createConnection();
277            } else {
278                return ((ConnectionFactory) connectionFactory).createConnection(key.getUserName(), key.getPassword());
279            }
280        } else {
281            throw new IllegalStateException("connectionFactory should implement javax.jms.ConnectionFactory");
282        }
283    }
284
285    public void start() {
286        LOG.debug("Staring the PooledConnectionFactory: create on start = {}", isCreateConnectionOnStartup());
287        stopped.set(false);
288        if (isCreateConnectionOnStartup()) {
289            try {
290                // warm the pool by creating a connection during startup
291                createConnection().close();
292            } catch (JMSException e) {
293                LOG.warn("Create pooled connection during start failed. This exception will be ignored.", e);
294            }
295        }
296    }
297
298    public void stop() {
299        if (stopped.compareAndSet(false, true)) {
300            LOG.debug("Stopping the PooledConnectionFactory, number of connections in cache: {}",
301                    connectionsPool != null ? connectionsPool.getNumActive() : 0);
302            try {
303                if (connectionsPool != null) {
304                    connectionsPool.close();
305                }
306            } catch (Exception e) {
307            }
308        }
309    }
310
311    /**
312     * Clears all connections from the pool.  Each connection that is currently in the pool is
313     * closed and removed from the pool.  A new connection will be created on the next call to
314     * {@link #createConnection}.  Care should be taken when using this method as Connections that
315     * are in use be client's will be closed.
316     */
317    public void clear() {
318
319        if (stopped.get()) {
320            return;
321        }
322
323        getConnectionsPool().clear();
324    }
325
326    /**
327     * Returns the currently configured maximum number of sessions a pooled Connection will
328     * create before it either blocks or throws an exception when a new session is requested,
329     * depending on configuration.
330     *
331     * @return the number of session instances that can be taken from a pooled connection.
332     */
333    public int getMaximumActiveSessionPerConnection() {
334        return maximumActiveSessionPerConnection;
335    }
336
337    /**
338     * Sets the maximum number of active sessions per connection
339     *
340     * @param maximumActiveSessionPerConnection
341     *      The maximum number of active session per connection in the pool.
342     */
343    public void setMaximumActiveSessionPerConnection(int maximumActiveSessionPerConnection) {
344        this.maximumActiveSessionPerConnection = maximumActiveSessionPerConnection;
345    }
346
347    /**
348     * Controls the behavior of the internal session pool. By default the call to
349     * Connection.getSession() will block if the session pool is full.  If the
350     * argument false is given, it will change the default behavior and instead the
351     * call to getSession() will throw a JMSException.
352     *
353     * The size of the session pool is controlled by the @see #maximumActive
354     * property.
355     *
356     * @param block - if true, the call to getSession() blocks if the pool is full
357     * until a session object is available.  defaults to true.
358     */
359    public void setBlockIfSessionPoolIsFull(boolean block) {
360        this.blockIfSessionPoolIsFull = block;
361    }
362
363    /**
364     * Returns whether a pooled Connection will enter a blocked state or will throw an Exception
365     * once the maximum number of sessions has been borrowed from the the Session Pool.
366     *
367     * @return true if the pooled Connection createSession method will block when the limit is hit.
368     * @see #setBlockIfSessionPoolIsFull(boolean)
369     */
370    public boolean isBlockIfSessionPoolIsFull() {
371        return this.blockIfSessionPoolIsFull;
372    }
373
374    /**
375     * Returns the maximum number to pooled Connections that this factory will allow before it
376     * begins to return connections from the pool on calls to ({@link #createConnection}.
377     *
378     * @return the maxConnections that will be created for this pool.
379     */
380    public int getMaxConnections() {
381        return getConnectionsPool().getMaxIdle();
382    }
383
384    /**
385     * Sets the maximum number of pooled Connections (defaults to one).  Each call to
386     * {@link #createConnection} will result in a new Connection being create up to the max
387     * connections value.
388     *
389     * @param maxConnections the maxConnections to set
390     */
391    public void setMaxConnections(int maxConnections) {
392        getConnectionsPool().setMaxIdle(maxConnections);
393    }
394
395    /**
396     * Gets the Idle timeout value applied to new Connection's that are created by this pool.
397     * <p/>
398     * The idle timeout is used determine if a Connection instance has sat to long in the pool unused
399     * and if so is closed and removed from the pool.  The default value is 30 seconds.
400     *
401     * @return idle timeout value (milliseconds)
402     */
403    public int getIdleTimeout() {
404        return idleTimeout;
405    }
406
407    /**
408     * Sets the idle timeout  value for Connection's that are created by this pool in Milliseconds,
409     * defaults to 30 seconds.
410     * <p/>
411     * For a Connection that is in the pool but has no current users the idle timeout determines how
412     * long the Connection can live before it is eligible for removal from the pool.  Normally the
413     * connections are tested when an attempt to check one out occurs so a Connection instance can sit
414     * in the pool much longer than its idle timeout if connections are used infrequently.
415     *
416     * @param idleTimeout
417     *      The maximum time a pooled Connection can sit unused before it is eligible for removal.
418     */
419    public void setIdleTimeout(int idleTimeout) {
420        this.idleTimeout = idleTimeout;
421    }
422
423    /**
424     * allow connections to expire, irrespective of load or idle time. This is useful with failover
425     * to force a reconnect from the pool, to reestablish load balancing or use of the master post recovery
426     *
427     * @param expiryTimeout non zero in milliseconds
428     */
429    public void setExpiryTimeout(long expiryTimeout) {
430        this.expiryTimeout = expiryTimeout;
431    }
432
433    /**
434     * @return the configured expiration timeout for connections in the pool.
435     */
436    public long getExpiryTimeout() {
437        return expiryTimeout;
438    }
439
440    /**
441     * @return true if a Connection is created immediately on a call to {@link start}.
442     */
443    public boolean isCreateConnectionOnStartup() {
444        return createConnectionOnStartup;
445    }
446
447    /**
448     * Whether to create a connection on starting this {@link PooledConnectionFactory}.
449     * <p/>
450     * This can be used to warm-up the pool on startup. Notice that any kind of exception
451     * happens during startup is logged at WARN level and ignored.
452     *
453     * @param createConnectionOnStartup <tt>true</tt> to create a connection on startup
454     */
455    public void setCreateConnectionOnStartup(boolean createConnectionOnStartup) {
456        this.createConnectionOnStartup = createConnectionOnStartup;
457    }
458
459    /**
460     * Should Sessions use one anonymous producer for all producer requests or should a new
461     * MessageProducer be created for each request to create a producer object, default is true.
462     *
463     * When enabled the session only needs to allocate one MessageProducer for all requests and
464     * the MessageProducer#send(destination, message) method can be used.  Normally this is the
465     * right thing to do however it does result in the Broker not showing the producers per
466     * destination.
467     *
468     * @return true if a PooledSession will use only a single anonymous message producer instance.
469     */
470    public boolean isUseAnonymousProducers() {
471        return this.useAnonymousProducers;
472    }
473
474    /**
475     * Sets whether a PooledSession uses only one anonymous MessageProducer instance or creates
476     * a new MessageProducer for each call the create a MessageProducer.
477     *
478     * @param value
479     *      Boolean value that configures whether anonymous producers are used.
480     */
481    public void setUseAnonymousProducers(boolean value) {
482        this.useAnonymousProducers = value;
483    }
484
485    /**
486     * Gets the Pool of ConnectionPool instances which are keyed by different ConnectionKeys.
487     *
488     * @return this factories pool of ConnectionPool instances.
489     */
490    protected GenericKeyedObjectPool<ConnectionKey, ConnectionPool> getConnectionsPool() {
491        initConnectionsPool();
492        return this.connectionsPool;
493    }
494
495    /**
496     * Sets the number of milliseconds to sleep between runs of the idle Connection eviction thread.
497     * When non-positive, no idle object eviction thread will be run, and Connections will only be
498     * checked on borrow to determine if they have sat idle for too long or have failed for some
499     * other reason.
500     * <p/>
501     * By default this value is set to -1 and no expiration thread ever runs.
502     *
503     * @param timeBetweenExpirationCheckMillis
504     *      The time to wait between runs of the idle Connection eviction thread.
505     */
506    public void setTimeBetweenExpirationCheckMillis(long timeBetweenExpirationCheckMillis) {
507        getConnectionsPool().setTimeBetweenEvictionRunsMillis(timeBetweenExpirationCheckMillis);
508    }
509
510    /**
511     * @return the number of milliseconds to sleep between runs of the idle connection eviction thread.
512     */
513    public long getTimeBetweenExpirationCheckMillis() {
514        return getConnectionsPool().getTimeBetweenEvictionRunsMillis();
515    }
516
517    /**
518     * @return the number of Connections currently in the Pool
519     */
520    public int getNumConnections() {
521        return getConnectionsPool().getNumIdle();
522    }
523
524    /**
525     * Delegate that creates each instance of an ConnectionPool object.  Subclasses can override
526     * this method to customize the type of connection pool returned.
527     *
528     * @param connection
529     *
530     * @return instance of a new ConnectionPool.
531     */
532    protected ConnectionPool createConnectionPool(Connection connection) {
533        return new ConnectionPool(connection);
534    }
535
536    /**
537     * Returns the timeout to use for blocking creating new sessions
538     *
539     * @return true if the pooled Connection createSession method will block when the limit is hit.
540     * @see #setBlockIfSessionPoolIsFull(boolean)
541     */
542    public long getBlockIfSessionPoolIsFullTimeout() {
543        return blockIfSessionPoolIsFullTimeout;
544    }
545
546    /**
547     * Controls the behavior of the internal session pool. By default the call to
548     * Connection.getSession() will block if the session pool is full.  This setting
549     * will affect how long it blocks and throws an exception after the timeout.
550     *
551     * The size of the session pool is controlled by the @see #maximumActive
552     * property.
553     *
554     * Whether or not the call to create session blocks is controlled by the @see #blockIfSessionPoolIsFull
555     * property
556     *
557     * @param blockIfSessionPoolIsFullTimeout - if blockIfSessionPoolIsFullTimeout is true,
558     *                                        then use this setting to configure how long to block before retry
559     */
560    public void setBlockIfSessionPoolIsFullTimeout(long blockIfSessionPoolIsFullTimeout) {
561        this.blockIfSessionPoolIsFullTimeout = blockIfSessionPoolIsFullTimeout;
562    }
563
564    /**
565     * @return true if the underlying connection will be renewed on JMSException, false otherwise
566     */
567    public boolean isReconnectOnException() {
568        return reconnectOnException;
569    }
570
571    /**
572     * Controls weather the underlying connection should be reset (and renewed) on JMSException
573     *
574     * @param reconnectOnException
575     *          Boolean value that configures whether reconnect on exception should happen
576     */
577    public void setReconnectOnException(boolean reconnectOnException) {
578        this.reconnectOnException = reconnectOnException;
579    }
580
581    /**
582     * Called by any superclass that implements a JNDIReferencable or similar that needs to collect
583     * the properties of this class for storage etc.
584     *
585     * This method should be updated any time there is a new property added.
586     *
587     * @param props
588     *        a properties object that should be filled in with this objects property values.
589     */
590    protected void populateProperties(Properties props) {
591        props.setProperty("maximumActiveSessionPerConnection", Integer.toString(getMaximumActiveSessionPerConnection()));
592        props.setProperty("maxConnections", Integer.toString(getMaxConnections()));
593        props.setProperty("idleTimeout", Integer.toString(getIdleTimeout()));
594        props.setProperty("expiryTimeout", Long.toString(getExpiryTimeout()));
595        props.setProperty("timeBetweenExpirationCheckMillis", Long.toString(getTimeBetweenExpirationCheckMillis()));
596        props.setProperty("createConnectionOnStartup", Boolean.toString(isCreateConnectionOnStartup()));
597        props.setProperty("useAnonymousProducers", Boolean.toString(isUseAnonymousProducers()));
598        props.setProperty("blockIfSessionPoolIsFullTimeout", Long.toString(getBlockIfSessionPoolIsFullTimeout()));
599        props.setProperty("reconnectOnException", Boolean.toString(isReconnectOnException()));
600    }
601}