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}