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 */
017
018package org.apache.activemq.jms.pool;
019
020import java.util.List;
021import java.util.concurrent.CopyOnWriteArrayList;
022import java.util.concurrent.atomic.AtomicBoolean;
023
024import javax.jms.Connection;
025import javax.jms.ExceptionListener;
026import javax.jms.IllegalStateException;
027import javax.jms.JMSException;
028import javax.jms.Session;
029import javax.jms.TemporaryQueue;
030import javax.jms.TemporaryTopic;
031
032import org.apache.commons.pool2.KeyedPooledObjectFactory;
033import org.apache.commons.pool2.PooledObject;
034import org.apache.commons.pool2.impl.DefaultPooledObject;
035import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
036import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
037import org.slf4j.Logger;
038import org.slf4j.LoggerFactory;
039
040/**
041 * Holds a real JMS connection along with the session pools associated with it.
042 * <p/>
043 * Instances of this class are shared amongst one or more PooledConnection object and must
044 * track the session objects that are loaned out for cleanup on close as well as ensuring
045 * that the temporary destinations of the managed Connection are purged when all references
046 * to this ConnectionPool are released.
047 */
048public class ConnectionPool implements ExceptionListener {
049    private static final transient Logger LOG = LoggerFactory.getLogger(ConnectionPool.class);
050
051    protected Connection connection;
052    private int referenceCount;
053    private long lastUsed = System.currentTimeMillis();
054    private final long firstUsed = lastUsed;
055    private boolean hasExpired;
056    private int idleTimeout = 30 * 1000;
057    private long expiryTimeout = 0l;
058    private boolean useAnonymousProducers = true;
059
060    private final AtomicBoolean started = new AtomicBoolean(false);
061    private final GenericKeyedObjectPool<SessionKey, SessionHolder> sessionPool;
062    private final List<PooledSession> loanedSessions = new CopyOnWriteArrayList<PooledSession>();
063    private boolean reconnectOnException;
064    private ExceptionListener parentExceptionListener;
065
066    public ConnectionPool(Connection connection) {
067        final GenericKeyedObjectPoolConfig poolConfig = new GenericKeyedObjectPoolConfig();
068        poolConfig.setJmxEnabled(false);
069        this.connection = wrap(connection);
070
071        // Create our internal Pool of session instances.
072        this.sessionPool = new GenericKeyedObjectPool<SessionKey, SessionHolder>(
073            new KeyedPooledObjectFactory<SessionKey, SessionHolder>() {
074                @Override
075                public PooledObject<SessionHolder> makeObject(SessionKey sessionKey) throws Exception {
076
077                    return new DefaultPooledObject<SessionHolder>(new SessionHolder(makeSession(sessionKey)));
078                }
079
080                @Override
081                public void destroyObject(SessionKey sessionKey, PooledObject<SessionHolder> pooledObject) throws Exception {
082                    ((SessionHolder)pooledObject.getObject()).close();
083                }
084
085                @Override
086                public boolean validateObject(SessionKey sessionKey, PooledObject<SessionHolder> pooledObject) {
087                    return true;
088                }
089
090                @Override
091                public void activateObject(SessionKey sessionKey, PooledObject<SessionHolder> pooledObject) throws Exception {
092                }
093
094                @Override
095                public void passivateObject(SessionKey sessionKey, PooledObject<SessionHolder> pooledObject) throws Exception {
096                }
097            }, poolConfig
098        );
099    }
100
101    // useful when external failure needs to force expiry
102    public void setHasExpired(boolean val) {
103        hasExpired = val;
104    }
105
106    protected Session makeSession(SessionKey key) throws JMSException {
107        return connection.createSession(key.isTransacted(), key.getAckMode());
108    }
109
110    protected Connection wrap(Connection connection) {
111        return connection;
112    }
113
114    protected void unWrap(Connection connection) {
115    }
116
117    public void start() throws JMSException {
118        if (started.compareAndSet(false, true)) {
119            try {
120                connection.start();
121            } catch (JMSException e) {
122                started.set(false);
123                throw(e);
124            }
125        }
126    }
127
128    public synchronized Connection getConnection() {
129        return connection;
130    }
131
132    public Session createSession(boolean transacted, int ackMode) throws JMSException {
133        SessionKey key = new SessionKey(transacted, ackMode);
134        PooledSession session;
135        try {
136            session = new PooledSession(key, sessionPool.borrowObject(key), sessionPool, key.isTransacted(), useAnonymousProducers);
137            session.addSessionEventListener(new PooledSessionEventListener() {
138
139                @Override
140                public void onTemporaryTopicCreate(TemporaryTopic tempTopic) {
141                }
142
143                @Override
144                public void onTemporaryQueueCreate(TemporaryQueue tempQueue) {
145                }
146
147                @Override
148                public void onSessionClosed(PooledSession session) {
149                    ConnectionPool.this.loanedSessions.remove(session);
150                }
151            });
152            this.loanedSessions.add(session);
153        } catch (Exception e) {
154            IllegalStateException illegalStateException = new IllegalStateException(e.toString());
155            illegalStateException.initCause(e);
156            throw illegalStateException;
157        }
158        return session;
159    }
160
161    public synchronized void close() {
162        if (connection != null) {
163            try {
164                sessionPool.close();
165            } catch (Exception e) {
166            } finally {
167                try {
168                    connection.close();
169                } catch (Exception e) {
170                } finally {
171                    connection = null;
172                }
173            }
174        }
175    }
176
177    public synchronized void incrementReferenceCount() {
178        referenceCount++;
179        lastUsed = System.currentTimeMillis();
180    }
181
182    public synchronized void decrementReferenceCount() {
183        referenceCount--;
184        lastUsed = System.currentTimeMillis();
185        if (referenceCount == 0) {
186            // Loaned sessions are those that are active in the sessionPool and
187            // have not been closed by the client before closing the connection.
188            // These need to be closed so that all session's reflect the fact
189            // that the parent Connection is closed.
190            for (PooledSession session : this.loanedSessions) {
191                try {
192                    session.close();
193                } catch (Exception e) {
194                }
195            }
196            this.loanedSessions.clear();
197
198            unWrap(getConnection());
199
200            expiredCheck();
201        }
202    }
203
204    /**
205     * Determines if this Connection has expired.
206     * <p/>
207     * A ConnectionPool is considered expired when all references to it are released AND either
208     * the configured idleTimeout has elapsed OR the configured expiryTimeout has elapsed.
209     * Once a ConnectionPool is determined to have expired its underlying Connection is closed.
210     *
211     * @return true if this connection has expired.
212     */
213    public synchronized boolean expiredCheck() {
214
215        boolean expired = false;
216
217        if (connection == null) {
218            return true;
219        }
220
221        if (hasExpired) {
222            if (referenceCount == 0) {
223                close();
224                expired = true;
225            }
226        }
227
228        if (expiryTimeout > 0 && System.currentTimeMillis() > firstUsed + expiryTimeout) {
229            hasExpired = true;
230            if (referenceCount == 0) {
231                close();
232                expired = true;
233            }
234        }
235
236        // Only set hasExpired here is no references, as a Connection with references is by
237        // definition not idle at this time.
238        if (referenceCount == 0 && idleTimeout > 0 && System.currentTimeMillis() > lastUsed + idleTimeout) {
239            hasExpired = true;
240            close();
241            expired = true;
242        }
243
244        return expired;
245    }
246
247    public int getIdleTimeout() {
248        return idleTimeout;
249    }
250
251    public void setIdleTimeout(int idleTimeout) {
252        this.idleTimeout = idleTimeout;
253    }
254
255    public void setExpiryTimeout(long expiryTimeout) {
256        this.expiryTimeout = expiryTimeout;
257    }
258
259    public long getExpiryTimeout() {
260        return expiryTimeout;
261    }
262
263    public int getMaximumActiveSessionPerConnection() {
264        return this.sessionPool.getMaxTotalPerKey();
265    }
266
267    public void setMaximumActiveSessionPerConnection(int maximumActiveSessionPerConnection) {
268        this.sessionPool.setMaxTotalPerKey(maximumActiveSessionPerConnection);
269    }
270
271    public boolean isUseAnonymousProducers() {
272        return this.useAnonymousProducers;
273    }
274
275    public void setUseAnonymousProducers(boolean value) {
276        this.useAnonymousProducers = value;
277    }
278
279    /**
280     * @return the total number of Pooled session including idle sessions that are not
281     *          currently loaned out to any client.
282     */
283    public int getNumSessions() {
284        return this.sessionPool.getNumIdle() + this.sessionPool.getNumActive();
285    }
286
287    /**
288     * @return the total number of Sessions that are in the Session pool but not loaned out.
289     */
290    public int getNumIdleSessions() {
291        return this.sessionPool.getNumIdle();
292    }
293
294    /**
295     * @return the total number of Session's that have been loaned to PooledConnection instances.
296     */
297    public int getNumActiveSessions() {
298        return this.sessionPool.getNumActive();
299    }
300
301    /**
302     * Configure whether the createSession method should block when there are no more idle sessions and the
303     * pool already contains the maximum number of active sessions.  If false the create method will fail
304     * and throw an exception.
305     *
306     * @param block
307     *          Indicates whether blocking should be used to wait for more space to create a session.
308     */
309    public void setBlockIfSessionPoolIsFull(boolean block) {
310        this.sessionPool.setBlockWhenExhausted(block);
311    }
312
313    public boolean isBlockIfSessionPoolIsFull() {
314        return this.sessionPool.getBlockWhenExhausted();
315    }
316
317    /**
318     * Returns the timeout to use for blocking creating new sessions
319     *
320     * @return true if the pooled Connection createSession method will block when the limit is hit.
321     * @see #setBlockIfSessionPoolIsFull(boolean)
322     */
323    public long getBlockIfSessionPoolIsFullTimeout() {
324        return this.sessionPool.getMaxWaitMillis();
325    }
326
327    /**
328     * Controls the behavior of the internal session pool. By default the call to
329     * Connection.getSession() will block if the session pool is full.  This setting
330     * will affect how long it blocks and throws an exception after the timeout.
331     *
332     * The size of the session pool is controlled by the @see #maximumActive
333     * property.
334     *
335     * Whether or not the call to create session blocks is controlled by the @see #blockIfSessionPoolIsFull
336     * property
337     *
338     * @param blockIfSessionPoolIsFullTimeout - if blockIfSessionPoolIsFullTimeout is true,
339     *                                        then use this setting to configure how long to block before retry
340     */
341    public void setBlockIfSessionPoolIsFullTimeout(long blockIfSessionPoolIsFullTimeout) {
342        this.sessionPool.setMaxWaitMillis(blockIfSessionPoolIsFullTimeout);
343    }
344
345    /**
346     * @return true if the underlying connection will be renewed on JMSException, false otherwise
347     */
348    public boolean isReconnectOnException() {
349        return reconnectOnException;
350    }
351
352    /**
353     * Controls weather the underlying connection should be reset (and renewed) on JMSException
354     *
355     * @param reconnectOnException
356     *          Boolean value that configures whether reconnect on exception should happen
357     */
358    public void setReconnectOnException(boolean reconnectOnException) {
359        this.reconnectOnException = reconnectOnException;
360        try {
361            if (isReconnectOnException()) {
362                if (connection.getExceptionListener() != null) {
363                    parentExceptionListener = connection.getExceptionListener();
364                }
365                connection.setExceptionListener(this);
366            } else {
367                if (parentExceptionListener != null) {
368                    connection.setExceptionListener(parentExceptionListener);
369                }
370                parentExceptionListener = null;
371            }
372        } catch (JMSException jmse) {
373            LOG.warn("Cannot set reconnect exception listener", jmse);
374        }
375    }
376
377    @Override
378    public void onException(JMSException exception) {
379        close();
380        if (parentExceptionListener != null) {
381            parentExceptionListener.onException(exception);
382        }
383    }
384
385    @Override
386    public String toString() {
387        return "ConnectionPool[" + connection + "]";
388    }
389}