001/* 002 * Copyright 2002-2017 the original author or authors. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * https://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016 017package org.springframework.jms.connection; 018 019import javax.jms.Connection; 020import javax.jms.ConnectionFactory; 021import javax.jms.JMSException; 022import javax.jms.Session; 023import javax.jms.TransactionRolledBackException; 024 025import org.springframework.beans.factory.InitializingBean; 026import org.springframework.transaction.CannotCreateTransactionException; 027import org.springframework.transaction.InvalidIsolationLevelException; 028import org.springframework.transaction.TransactionDefinition; 029import org.springframework.transaction.TransactionSystemException; 030import org.springframework.transaction.UnexpectedRollbackException; 031import org.springframework.transaction.support.AbstractPlatformTransactionManager; 032import org.springframework.transaction.support.DefaultTransactionStatus; 033import org.springframework.transaction.support.ResourceTransactionManager; 034import org.springframework.transaction.support.SmartTransactionObject; 035import org.springframework.transaction.support.TransactionSynchronizationManager; 036 037/** 038 * {@link org.springframework.transaction.PlatformTransactionManager} implementation 039 * for a single JMS {@link javax.jms.ConnectionFactory}. Binds a JMS 040 * Connection/Session pair from the specified ConnectionFactory to the thread, 041 * potentially allowing for one thread-bound Session per ConnectionFactory. 042 * 043 * <p>This local strategy is an alternative to executing JMS operations within 044 * JTA transactions. Its advantage is that it is able to work in any environment, 045 * for example a standalone application or a test suite, with any message broker 046 * as target. However, this strategy is <i>not</i> able to provide XA transactions, 047 * for example in order to share transactions between messaging and database access. 048 * A full JTA/XA setup is required for XA transactions, typically using Spring's 049 * {@link org.springframework.transaction.jta.JtaTransactionManager} as strategy. 050 * 051 * <p>Application code is required to retrieve the transactional JMS Session via 052 * {@link ConnectionFactoryUtils#getTransactionalSession} instead of a standard 053 * Java EE-style {@link ConnectionFactory#createConnection()} call with subsequent 054 * Session creation. Spring's {@link org.springframework.jms.core.JmsTemplate} 055 * will autodetect a thread-bound Session and automatically participate in it. 056 * 057 * <p>Alternatively, you can allow application code to work with the standard 058 * Java EE-style lookup pattern on a ConnectionFactory, for example for legacy code 059 * that is not aware of Spring at all. In that case, define a 060 * {@link TransactionAwareConnectionFactoryProxy} for your target ConnectionFactory, 061 * which will automatically participate in Spring-managed transactions. 062 * 063 * <p><b>The use of {@link CachingConnectionFactory} as a target for this 064 * transaction manager is strongly recommended.</b> CachingConnectionFactory 065 * uses a single JMS Connection for all JMS access in order to avoid the overhead 066 * of repeated Connection creation, as well as maintaining a cache of Sessions. 067 * Each transaction will then share the same JMS Connection, while still using 068 * its own individual JMS Session. 069 * 070 * <p>The use of a <i>raw</i> target ConnectionFactory would not only be inefficient 071 * because of the lack of resource reuse. It might also lead to strange effects 072 * when your JMS driver doesn't accept {@code MessageProducer.close()} calls 073 * and/or {@code MessageConsumer.close()} calls before {@code Session.commit()}, 074 * with the latter supposed to commit all the messages that have been sent through the 075 * producer handle and received through the consumer handle. As a safe general solution, 076 * always pass in a {@link CachingConnectionFactory} into this transaction manager's 077 * {@link #setConnectionFactory "connectionFactory"} property. 078 * 079 * <p>Transaction synchronization is turned off by default, as this manager might 080 * be used alongside a datastore-based Spring transaction manager such as the 081 * JDBC {@link org.springframework.jdbc.datasource.DataSourceTransactionManager}, 082 * which has stronger needs for synchronization. 083 * 084 * @author Juergen Hoeller 085 * @since 1.1 086 * @see ConnectionFactoryUtils#getTransactionalSession 087 * @see TransactionAwareConnectionFactoryProxy 088 * @see org.springframework.jms.core.JmsTemplate 089 */ 090@SuppressWarnings("serial") 091public class JmsTransactionManager extends AbstractPlatformTransactionManager 092 implements ResourceTransactionManager, InitializingBean { 093 094 private ConnectionFactory connectionFactory; 095 096 097 /** 098 * Create a new JmsTransactionManager for bean-style usage. 099 * <p>Note: The ConnectionFactory has to be set before using the instance. 100 * This constructor can be used to prepare a JmsTemplate via a BeanFactory, 101 * typically setting the ConnectionFactory via setConnectionFactory. 102 * <p>Turns off transaction synchronization by default, as this manager might 103 * be used alongside a datastore-based Spring transaction manager like 104 * DataSourceTransactionManager, which has stronger needs for synchronization. 105 * Only one manager is allowed to drive synchronization at any point of time. 106 * @see #setConnectionFactory 107 * @see #setTransactionSynchronization 108 */ 109 public JmsTransactionManager() { 110 setTransactionSynchronization(SYNCHRONIZATION_NEVER); 111 } 112 113 /** 114 * Create a new JmsTransactionManager, given a ConnectionFactory. 115 * @param connectionFactory the ConnectionFactory to obtain connections from 116 */ 117 public JmsTransactionManager(ConnectionFactory connectionFactory) { 118 this(); 119 setConnectionFactory(connectionFactory); 120 afterPropertiesSet(); 121 } 122 123 124 /** 125 * Set the JMS ConnectionFactory that this instance should manage transactions for. 126 */ 127 public void setConnectionFactory(ConnectionFactory cf) { 128 if (cf instanceof TransactionAwareConnectionFactoryProxy) { 129 // If we got a TransactionAwareConnectionFactoryProxy, we need to perform transactions 130 // for its underlying target ConnectionFactory, else JMS access code won't see 131 // properly exposed transactions (i.e. transactions for the target ConnectionFactory). 132 this.connectionFactory = ((TransactionAwareConnectionFactoryProxy) cf).getTargetConnectionFactory(); 133 } 134 else { 135 this.connectionFactory = cf; 136 } 137 } 138 139 /** 140 * Return the JMS ConnectionFactory that this instance should manage transactions for. 141 */ 142 public ConnectionFactory getConnectionFactory() { 143 return this.connectionFactory; 144 } 145 146 /** 147 * Make sure the ConnectionFactory has been set. 148 */ 149 @Override 150 public void afterPropertiesSet() { 151 if (getConnectionFactory() == null) { 152 throw new IllegalArgumentException("Property 'connectionFactory' is required"); 153 } 154 } 155 156 157 @Override 158 public Object getResourceFactory() { 159 return getConnectionFactory(); 160 } 161 162 @Override 163 protected Object doGetTransaction() { 164 JmsTransactionObject txObject = new JmsTransactionObject(); 165 txObject.setResourceHolder( 166 (JmsResourceHolder) TransactionSynchronizationManager.getResource(getConnectionFactory())); 167 return txObject; 168 } 169 170 @Override 171 protected boolean isExistingTransaction(Object transaction) { 172 JmsTransactionObject txObject = (JmsTransactionObject) transaction; 173 return txObject.hasResourceHolder(); 174 } 175 176 @Override 177 protected void doBegin(Object transaction, TransactionDefinition definition) { 178 if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) { 179 throw new InvalidIsolationLevelException("JMS does not support an isolation level concept"); 180 } 181 182 JmsTransactionObject txObject = (JmsTransactionObject) transaction; 183 Connection con = null; 184 Session session = null; 185 try { 186 con = createConnection(); 187 session = createSession(con); 188 if (logger.isDebugEnabled()) { 189 logger.debug("Created JMS transaction on Session [" + session + "] from Connection [" + con + "]"); 190 } 191 txObject.setResourceHolder(new JmsResourceHolder(getConnectionFactory(), con, session)); 192 txObject.getResourceHolder().setSynchronizedWithTransaction(true); 193 int timeout = determineTimeout(definition); 194 if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) { 195 txObject.getResourceHolder().setTimeoutInSeconds(timeout); 196 } 197 TransactionSynchronizationManager.bindResource(getConnectionFactory(), txObject.getResourceHolder()); 198 } 199 catch (Throwable ex) { 200 if (session != null) { 201 try { 202 session.close(); 203 } 204 catch (Throwable ex2) { 205 // ignore 206 } 207 } 208 if (con != null) { 209 try { 210 con.close(); 211 } 212 catch (Throwable ex2) { 213 // ignore 214 } 215 } 216 throw new CannotCreateTransactionException("Could not create JMS transaction", ex); 217 } 218 } 219 220 @Override 221 protected Object doSuspend(Object transaction) { 222 JmsTransactionObject txObject = (JmsTransactionObject) transaction; 223 txObject.setResourceHolder(null); 224 return TransactionSynchronizationManager.unbindResource(getConnectionFactory()); 225 } 226 227 @Override 228 protected void doResume(Object transaction, Object suspendedResources) { 229 TransactionSynchronizationManager.bindResource(getConnectionFactory(), suspendedResources); 230 } 231 232 @Override 233 protected void doCommit(DefaultTransactionStatus status) { 234 JmsTransactionObject txObject = (JmsTransactionObject) status.getTransaction(); 235 Session session = txObject.getResourceHolder().getSession(); 236 try { 237 if (status.isDebug()) { 238 logger.debug("Committing JMS transaction on Session [" + session + "]"); 239 } 240 session.commit(); 241 } 242 catch (TransactionRolledBackException ex) { 243 throw new UnexpectedRollbackException("JMS transaction rolled back", ex); 244 } 245 catch (JMSException ex) { 246 throw new TransactionSystemException("Could not commit JMS transaction", ex); 247 } 248 } 249 250 @Override 251 protected void doRollback(DefaultTransactionStatus status) { 252 JmsTransactionObject txObject = (JmsTransactionObject) status.getTransaction(); 253 Session session = txObject.getResourceHolder().getSession(); 254 try { 255 if (status.isDebug()) { 256 logger.debug("Rolling back JMS transaction on Session [" + session + "]"); 257 } 258 session.rollback(); 259 } 260 catch (JMSException ex) { 261 throw new TransactionSystemException("Could not roll back JMS transaction", ex); 262 } 263 } 264 265 @Override 266 protected void doSetRollbackOnly(DefaultTransactionStatus status) { 267 JmsTransactionObject txObject = (JmsTransactionObject) status.getTransaction(); 268 txObject.getResourceHolder().setRollbackOnly(); 269 } 270 271 @Override 272 protected void doCleanupAfterCompletion(Object transaction) { 273 JmsTransactionObject txObject = (JmsTransactionObject) transaction; 274 TransactionSynchronizationManager.unbindResource(getConnectionFactory()); 275 txObject.getResourceHolder().closeAll(); 276 txObject.getResourceHolder().clear(); 277 } 278 279 280 /** 281 * Create a JMS Connection via this template's ConnectionFactory. 282 * <p>This implementation uses JMS 1.1 API. 283 * @return the new JMS Connection 284 * @throws javax.jms.JMSException if thrown by JMS API methods 285 */ 286 protected Connection createConnection() throws JMSException { 287 return getConnectionFactory().createConnection(); 288 } 289 290 /** 291 * Create a JMS Session for the given Connection. 292 * <p>This implementation uses JMS 1.1 API. 293 * @param con the JMS Connection to create a Session for 294 * @return the new JMS Session 295 * @throws javax.jms.JMSException if thrown by JMS API methods 296 */ 297 protected Session createSession(Connection con) throws JMSException { 298 return con.createSession(true, Session.AUTO_ACKNOWLEDGE); 299 } 300 301 302 /** 303 * JMS transaction object, representing a JmsResourceHolder. 304 * Used as transaction object by JmsTransactionManager. 305 * @see JmsResourceHolder 306 */ 307 private static class JmsTransactionObject implements SmartTransactionObject { 308 309 private JmsResourceHolder resourceHolder; 310 311 public void setResourceHolder(JmsResourceHolder resourceHolder) { 312 this.resourceHolder = resourceHolder; 313 } 314 315 public JmsResourceHolder getResourceHolder() { 316 return this.resourceHolder; 317 } 318 319 public boolean hasResourceHolder() { 320 return (this.resourceHolder != null); 321 } 322 323 @Override 324 public boolean isRollbackOnly() { 325 return this.resourceHolder.isRollbackOnly(); 326 } 327 328 @Override 329 public void flush() { 330 // no-op 331 } 332 } 333 334}