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}