Release shared JMS Connection on Single/CachingConnectionFactory.stop()

See gh-30612

Author: jhoeller

spring-jms/src/main/java/org/springframework/jms/connection/SingleConnectionFactory.java

@@ -40,6 +40,7 @@
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.Lifecycle;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -85,7 +86,7 @@
  * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setCacheLevel
  */
 public class SingleConnectionFactory implements ConnectionFactory, QueueConnectionFactory,
-		TopicConnectionFactory, ExceptionListener, InitializingBean, DisposableBean {
+		TopicConnectionFactory, ExceptionListener, InitializingBean, DisposableBean, Lifecycle {
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -330,6 +331,67 @@ protected Connection getConnection() throws JMSException {
 		}
 	}
 
+	/**
+	 * Exception listener callback that renews the underlying single Connection.
+	 * @see #resetConnection()
+	 */
+	@Override
+	public void onException(JMSException ex) {
+		logger.info("Encountered a JMSException - resetting the underlying JMS Connection", ex);
+		resetConnection();
+	}
+
+	/**
+	 * Close the underlying shared connection.
+	 * The provider of this ConnectionFactory needs to care for proper shutdown.
+	 * <p>As this bean implements DisposableBean, a bean factory will
+	 * automatically invoke this on destruction of its cached singletons.
+	 * @see #resetConnection()
+	 */
+	@Override
+	public void destroy() {
+		resetConnection();
+	}
+
+	/**
+	 * Initialize the underlying shared connection on start.
+	 * @since 6.1
+	 * @see #initConnection()
+	 */
+	@Override
+	public void start() {
+		try {
+			initConnection();
+		}
+		catch (JMSException ex) {
+			logger.info("Start attempt failed for shared JMS Connection", ex);
+		}
+	}
+
+	/**
+	 * Reset the underlying shared connection on stop.
+	 * @since 6.1
+	 * @see #resetConnection()
+	 */
+	@Override
+	public void stop() {
+		resetConnection();
+	}
+
+	/**
+	 * Check whether there is currently an underlying connection.
+	 * @since 6.1
+	 * @see #start()
+	 * @see #stop()
+	 */
+	@Override
+	public boolean isRunning() {
+		synchronized (this.connectionMonitor) {
+			return (this.connection != null);
+		}
+	}
+
+
 	/**
 	 * Initialize the underlying shared Connection.
 	 * <p>Closes and reinitializes the Connection if an underlying
@@ -375,41 +437,6 @@ public void initConnection() throws JMSException {
 		}
 	}
 
-	/**
-	 * Exception listener callback that renews the underlying single Connection.
-	 * @see #resetConnection()
-	 */
-	@Override
-	public void onException(JMSException ex) {
-		logger.info("Encountered a JMSException - resetting the underlying JMS Connection", ex);
-		resetConnection();
-	}
-
-	/**
-	 * Close the underlying shared connection.
-	 * The provider of this ConnectionFactory needs to care for proper shutdown.
-	 * <p>As this bean implements DisposableBean, a bean factory will
-	 * automatically invoke this on destruction of its cached singletons.
-	 * @see #resetConnection()
-	 */
-	@Override
-	public void destroy() {
-		resetConnection();
-	}
-
-	/**
-	 * Reset the underlying shared Connection, to be reinitialized on next access.
-	 * @see #closeConnection
-	 */
-	public void resetConnection() {
-		synchronized (this.connectionMonitor) {
-			if (this.connection != null) {
-				closeConnection(this.connection);
-			}
-			this.connection = null;
-		}
-	}
-
 	/**
 	 * Create a JMS Connection via this template's ConnectionFactory.
 	 * @return the new JMS Connection
@@ -501,6 +528,19 @@ else if (Boolean.TRUE.equals(this.pubSubMode) && con instanceof TopicConnection
 		}
 	}
 
+	/**
+	 * Reset the underlying shared Connection, to be reinitialized on next access.
+	 * @see #closeConnection
+	 */
+	public void resetConnection() {
+		synchronized (this.connectionMonitor) {
+			if (this.connection != null) {
+				closeConnection(this.connection);
+			}
+			this.connection = null;
+		}
+	}
+
 	/**
 	 * Close the given Connection.
 	 * @param con the Connection to close

spring-jms/src/test/java/org/springframework/jms/connection/SingleConnectionFactoryTests.java

@@ -367,13 +367,12 @@ public void setExceptionListener(ExceptionListener exceptionListener) throws JMS
 		// Prepare base JMS ConnectionFactory
 		// - createConnection(1st) -> TestConnection,
 		// - createConnection(2nd and next) -> FailingTestConnection
-		TestConnection testCon = new TestConnection();
 		FailingTestConnection failingCon = new FailingTestConnection();
 		AtomicInteger createConnectionMethodCounter = new AtomicInteger();
 		ConnectionFactory cf = mock(ConnectionFactory.class);
 		given(cf.createConnection()).willAnswer(invocation -> {
 			int methodInvocationCounter = createConnectionMethodCounter.incrementAndGet();
-			return methodInvocationCounter == 1 ? testCon : failingCon;
+			return (methodInvocationCounter >= 4 ? failingCon : new TestConnection());
 		});
 
 		// Prepare SingleConnectionFactory (setReconnectOnException())
@@ -382,18 +381,43 @@ public void setExceptionListener(ExceptionListener exceptionListener) throws JMS
 		scf.setReconnectOnException(true);
 		Field conField = ReflectionUtils.findField(SingleConnectionFactory.class, "connection");
 		conField.setAccessible(true);
+		assertThat(scf.isRunning()).isFalse();
 
 		// Get connection (1st)
 		Connection con1 = scf.getConnection();
 		assertThat(createConnectionMethodCounter.get()).isEqualTo(1);
-		assertThat(con1).isNotNull();
 		assertThat(con1.getExceptionListener()).isNotNull();
-		assertThat(con1).isSameAs(testCon);
+		assertThat(con1).isSameAs(conField.get(scf));
+		assertThat(scf.isRunning()).isTrue();
+
 		// Get connection again, the same should be returned (shared connection till some problem)
 		Connection con2 = scf.getConnection();
 		assertThat(createConnectionMethodCounter.get()).isEqualTo(1);
 		assertThat(con2.getExceptionListener()).isNotNull();
 		assertThat(con2).isSameAs(con1);
+		assertThat(scf.isRunning()).isTrue();
+
+		// Explicit stop should reset connection
+		scf.stop();
+		assertThat(conField.get(scf)).isNull();
+		assertThat(scf.isRunning()).isFalse();
+		Connection con3 = scf.getConnection();
+		assertThat(createConnectionMethodCounter.get()).isEqualTo(2);
+		assertThat(con3.getExceptionListener()).isNotNull();
+		assertThat(con3).isNotSameAs(con2);
+		assertThat(scf.isRunning()).isTrue();
+
+		// Explicit stop-and-restart should refresh connection
+		scf.stop();
+		assertThat(conField.get(scf)).isNull();
+		assertThat(scf.isRunning()).isFalse();
+		scf.start();
+		assertThat(scf.isRunning()).isTrue();
+		assertThat(conField.get(scf)).isNotNull();
+		Connection con4 = scf.getConnection();
+		assertThat(createConnectionMethodCounter.get()).isEqualTo(3);
+		assertThat(con4.getExceptionListener()).isNotNull();
+		assertThat(con4).isNotSameAs(con3);
 
 		// Invoke reset connection to simulate problem with connection
 		// - SCF exception listener should be invoked -> connection should be set to null
@@ -405,16 +429,21 @@ public void setExceptionListener(ExceptionListener exceptionListener) throws JMS
 		// - JMSException should be returned from FailingTestConnection
 		// - connection should be still null (no new connection without exception listener like before fix)
 		assertThatExceptionOfType(JMSException.class).isThrownBy(() -> scf.getConnection());
-		assertThat(createConnectionMethodCounter.get()).isEqualTo(2);
+		assertThat(createConnectionMethodCounter.get()).isEqualTo(4);
 		assertThat(conField.get(scf)).isNull();
 
 		// Attempt to get connection again -> FailingTestConnection should be returned
-		//  - no JMSException is thrown, exception listener should be present
-		Connection con3 = scf.getConnection();
-		assertThat(createConnectionMethodCounter.get()).isEqualTo(3);
-		assertThat(con3).isNotNull();
-		assertThat(con3).isSameAs(failingCon);
-		assertThat(con3.getExceptionListener()).isNotNull();
+		// - no JMSException is thrown, exception listener should be present
+		Connection con5 = scf.getConnection();
+		assertThat(createConnectionMethodCounter.get()).isEqualTo(5);
+		assertThat(con5).isNotNull();
+		assertThat(con5).isSameAs(failingCon);
+		assertThat(con5.getExceptionListener()).isNotNull();
+		assertThat(con5).isNotSameAs(con4);
+
+		scf.destroy();
+		assertThat(conField.get(scf)).isNull();
+		assertThat(scf.isRunning()).isFalse();
 	}
 
 	@Test