001/* 002 * Copyright 2002-2019 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 java.lang.reflect.Method; 020import java.util.HashMap; 021import java.util.LinkedList; 022import java.util.Map; 023 024import javax.jms.Connection; 025import javax.jms.ConnectionFactory; 026import javax.jms.JMSException; 027import javax.jms.Session; 028import javax.jms.TransactionInProgressException; 029 030import org.apache.commons.logging.Log; 031import org.apache.commons.logging.LogFactory; 032 033import org.springframework.lang.Nullable; 034import org.springframework.transaction.support.ResourceHolderSupport; 035import org.springframework.transaction.support.TransactionSynchronizationManager; 036import org.springframework.util.Assert; 037import org.springframework.util.CollectionUtils; 038import org.springframework.util.ReflectionUtils; 039 040/** 041 * Resource holder wrapping a JMS {@link Connection} and a JMS {@link Session}. 042 * {@link JmsTransactionManager} binds instances of this class to the thread, 043 * for a given JMS {@link ConnectionFactory}. 044 * 045 * <p>Note: This is an SPI class, not intended to be used by applications. 046 * 047 * @author Juergen Hoeller 048 * @since 1.1 049 * @see JmsTransactionManager 050 * @see org.springframework.jms.core.JmsTemplate 051 */ 052public class JmsResourceHolder extends ResourceHolderSupport { 053 054 private static final Log logger = LogFactory.getLog(JmsResourceHolder.class); 055 056 @Nullable 057 private ConnectionFactory connectionFactory; 058 059 private boolean frozen = false; 060 061 private final LinkedList<Connection> connections = new LinkedList<>(); 062 063 private final LinkedList<Session> sessions = new LinkedList<>(); 064 065 private final Map<Connection, LinkedList<Session>> sessionsPerConnection = new HashMap<>(); 066 067 068 /** 069 * Create a new JmsResourceHolder that is open for resources to be added. 070 * @see #addConnection 071 * @see #addSession 072 */ 073 public JmsResourceHolder() { 074 } 075 076 /** 077 * Create a new JmsResourceHolder that is open for resources to be added. 078 * @param connectionFactory the JMS ConnectionFactory that this 079 * resource holder is associated with (may be {@code null}) 080 */ 081 public JmsResourceHolder(@Nullable ConnectionFactory connectionFactory) { 082 this.connectionFactory = connectionFactory; 083 } 084 085 /** 086 * Create a new JmsResourceHolder for the given JMS Session. 087 * @param session the JMS Session 088 */ 089 public JmsResourceHolder(Session session) { 090 addSession(session); 091 this.frozen = true; 092 } 093 094 /** 095 * Create a new JmsResourceHolder for the given JMS resources. 096 * @param connection the JMS Connection 097 * @param session the JMS Session 098 */ 099 public JmsResourceHolder(Connection connection, Session session) { 100 addConnection(connection); 101 addSession(session, connection); 102 this.frozen = true; 103 } 104 105 /** 106 * Create a new JmsResourceHolder for the given JMS resources. 107 * @param connectionFactory the JMS ConnectionFactory that this 108 * resource holder is associated with (may be {@code null}) 109 * @param connection the JMS Connection 110 * @param session the JMS Session 111 */ 112 public JmsResourceHolder(@Nullable ConnectionFactory connectionFactory, Connection connection, Session session) { 113 this.connectionFactory = connectionFactory; 114 addConnection(connection); 115 addSession(session, connection); 116 this.frozen = true; 117 } 118 119 120 /** 121 * Return whether this resource holder is frozen, i.e. does not 122 * allow for adding further Connections and Sessions to it. 123 * @see #addConnection 124 * @see #addSession 125 */ 126 public final boolean isFrozen() { 127 return this.frozen; 128 } 129 130 /** 131 * Add the given Connection to this resource holder. 132 */ 133 public final void addConnection(Connection connection) { 134 Assert.isTrue(!this.frozen, "Cannot add Connection because JmsResourceHolder is frozen"); 135 Assert.notNull(connection, "Connection must not be null"); 136 if (!this.connections.contains(connection)) { 137 this.connections.add(connection); 138 } 139 } 140 141 /** 142 * Add the given Session to this resource holder. 143 */ 144 public final void addSession(Session session) { 145 addSession(session, null); 146 } 147 148 /** 149 * Add the given Session to this resource holder, 150 * registered for a specific Connection. 151 */ 152 public final void addSession(Session session, @Nullable Connection connection) { 153 Assert.isTrue(!this.frozen, "Cannot add Session because JmsResourceHolder is frozen"); 154 Assert.notNull(session, "Session must not be null"); 155 if (!this.sessions.contains(session)) { 156 this.sessions.add(session); 157 if (connection != null) { 158 LinkedList<Session> sessions = 159 this.sessionsPerConnection.computeIfAbsent(connection, k -> new LinkedList<>()); 160 sessions.add(session); 161 } 162 } 163 } 164 165 /** 166 * Determine whether the given Session is registered 167 * with this resource holder. 168 */ 169 public boolean containsSession(Session session) { 170 return this.sessions.contains(session); 171 } 172 173 174 /** 175 * Return this resource holder's default Connection, 176 * or {@code null} if none. 177 */ 178 @Nullable 179 public Connection getConnection() { 180 return this.connections.peek(); 181 } 182 183 /** 184 * Return this resource holder's Connection of the given type, 185 * or {@code null} if none. 186 */ 187 @Nullable 188 public <C extends Connection> C getConnection(Class<C> connectionType) { 189 return CollectionUtils.findValueOfType(this.connections, connectionType); 190 } 191 192 /** 193 * Return an existing original Session, if any. 194 * <p>In contrast to {@link #getSession()}, this must not lazily initialize 195 * a new Session, not even in {@link JmsResourceHolder} subclasses. 196 */ 197 @Nullable 198 Session getOriginalSession() { 199 return this.sessions.peek(); 200 } 201 202 /** 203 * Return this resource holder's default Session, 204 * or {@code null} if none. 205 */ 206 @Nullable 207 public Session getSession() { 208 return this.sessions.peek(); 209 } 210 211 /** 212 * Return this resource holder's Session of the given type, 213 * or {@code null} if none. 214 */ 215 @Nullable 216 public <S extends Session> S getSession(Class<S> sessionType) { 217 return getSession(sessionType, null); 218 } 219 220 /** 221 * Return this resource holder's Session of the given type 222 * for the given connection, or {@code null} if none. 223 */ 224 @Nullable 225 public <S extends Session> S getSession(Class<S> sessionType, @Nullable Connection connection) { 226 LinkedList<Session> sessions = 227 (connection != null ? this.sessionsPerConnection.get(connection) : this.sessions); 228 return CollectionUtils.findValueOfType(sessions, sessionType); 229 } 230 231 232 /** 233 * Commit all of this resource holder's Sessions. 234 * @throws JMSException if thrown from a Session commit attempt 235 * @see Session#commit() 236 */ 237 public void commitAll() throws JMSException { 238 for (Session session : this.sessions) { 239 try { 240 session.commit(); 241 } 242 catch (TransactionInProgressException ex) { 243 // Ignore -> can only happen in case of a JTA transaction. 244 } 245 catch (javax.jms.IllegalStateException ex) { 246 if (this.connectionFactory != null) { 247 try { 248 Method getDataSourceMethod = this.connectionFactory.getClass().getMethod("getDataSource"); 249 Object ds = ReflectionUtils.invokeMethod(getDataSourceMethod, this.connectionFactory); 250 while (ds != null) { 251 if (TransactionSynchronizationManager.hasResource(ds)) { 252 // IllegalStateException from sharing the underlying JDBC Connection 253 // which typically gets committed first, e.g. with Oracle AQ --> ignore 254 return; 255 } 256 try { 257 // Check for decorated DataSource a la Spring's DelegatingDataSource 258 Method getTargetDataSourceMethod = ds.getClass().getMethod("getTargetDataSource"); 259 ds = ReflectionUtils.invokeMethod(getTargetDataSourceMethod, ds); 260 } 261 catch (NoSuchMethodException nsme) { 262 ds = null; 263 } 264 } 265 } 266 catch (Throwable ex2) { 267 if (logger.isDebugEnabled()) { 268 logger.debug("No working getDataSource method found on ConnectionFactory: " + ex2); 269 } 270 // No working getDataSource method - cannot perform DataSource transaction check 271 } 272 } 273 throw ex; 274 } 275 } 276 } 277 278 /** 279 * Close all of this resource holder's Sessions and clear its state. 280 * @see Session#close() 281 */ 282 public void closeAll() { 283 for (Session session : this.sessions) { 284 try { 285 session.close(); 286 } 287 catch (Throwable ex) { 288 logger.debug("Could not close synchronized JMS Session after transaction", ex); 289 } 290 } 291 for (Connection con : this.connections) { 292 ConnectionFactoryUtils.releaseConnection(con, this.connectionFactory, true); 293 } 294 this.connections.clear(); 295 this.sessions.clear(); 296 this.sessionsPerConnection.clear(); 297 } 298 299}