Release shared JMS Connection on Single/CachingConnectionFactory.stop()
See gh-30612
This commit is contained in:
@@ -40,6 +40,7 @@ import org.apache.commons.logging.LogFactory;
|
|||||||
|
|
||||||
import org.springframework.beans.factory.DisposableBean;
|
import org.springframework.beans.factory.DisposableBean;
|
||||||
import org.springframework.beans.factory.InitializingBean;
|
import org.springframework.beans.factory.InitializingBean;
|
||||||
|
import org.springframework.context.Lifecycle;
|
||||||
import org.springframework.lang.Nullable;
|
import org.springframework.lang.Nullable;
|
||||||
import org.springframework.util.Assert;
|
import org.springframework.util.Assert;
|
||||||
import org.springframework.util.ClassUtils;
|
import org.springframework.util.ClassUtils;
|
||||||
@@ -85,7 +86,7 @@ import org.springframework.util.ObjectUtils;
|
|||||||
* @see org.springframework.jms.listener.DefaultMessageListenerContainer#setCacheLevel
|
* @see org.springframework.jms.listener.DefaultMessageListenerContainer#setCacheLevel
|
||||||
*/
|
*/
|
||||||
public class SingleConnectionFactory implements ConnectionFactory, QueueConnectionFactory,
|
public class SingleConnectionFactory implements ConnectionFactory, QueueConnectionFactory,
|
||||||
TopicConnectionFactory, ExceptionListener, InitializingBean, DisposableBean {
|
TopicConnectionFactory, ExceptionListener, InitializingBean, DisposableBean, Lifecycle {
|
||||||
|
|
||||||
protected final Log logger = LogFactory.getLog(getClass());
|
protected final Log logger = LogFactory.getLog(getClass());
|
||||||
|
|
||||||
@@ -330,6 +331,67 @@ public class SingleConnectionFactory implements ConnectionFactory, QueueConnecti
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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.
|
* Initialize the underlying shared Connection.
|
||||||
* <p>Closes and reinitializes the Connection if an underlying
|
* <p>Closes and reinitializes the Connection if an underlying
|
||||||
@@ -375,41 +437,6 @@ public class SingleConnectionFactory implements ConnectionFactory, QueueConnecti
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* 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.
|
* Create a JMS Connection via this template's ConnectionFactory.
|
||||||
* @return the new JMS Connection
|
* @return the new JMS Connection
|
||||||
@@ -501,6 +528,19 @@ public class SingleConnectionFactory implements ConnectionFactory, QueueConnecti
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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.
|
* Close the given Connection.
|
||||||
* @param con the Connection to close
|
* @param con the Connection to close
|
||||||
|
|||||||
@@ -367,13 +367,12 @@ public class SingleConnectionFactoryTests {
|
|||||||
// Prepare base JMS ConnectionFactory
|
// Prepare base JMS ConnectionFactory
|
||||||
// - createConnection(1st) -> TestConnection,
|
// - createConnection(1st) -> TestConnection,
|
||||||
// - createConnection(2nd and next) -> FailingTestConnection
|
// - createConnection(2nd and next) -> FailingTestConnection
|
||||||
TestConnection testCon = new TestConnection();
|
|
||||||
FailingTestConnection failingCon = new FailingTestConnection();
|
FailingTestConnection failingCon = new FailingTestConnection();
|
||||||
AtomicInteger createConnectionMethodCounter = new AtomicInteger();
|
AtomicInteger createConnectionMethodCounter = new AtomicInteger();
|
||||||
ConnectionFactory cf = mock(ConnectionFactory.class);
|
ConnectionFactory cf = mock(ConnectionFactory.class);
|
||||||
given(cf.createConnection()).willAnswer(invocation -> {
|
given(cf.createConnection()).willAnswer(invocation -> {
|
||||||
int methodInvocationCounter = createConnectionMethodCounter.incrementAndGet();
|
int methodInvocationCounter = createConnectionMethodCounter.incrementAndGet();
|
||||||
return methodInvocationCounter == 1 ? testCon : failingCon;
|
return (methodInvocationCounter >= 4 ? failingCon : new TestConnection());
|
||||||
});
|
});
|
||||||
|
|
||||||
// Prepare SingleConnectionFactory (setReconnectOnException())
|
// Prepare SingleConnectionFactory (setReconnectOnException())
|
||||||
@@ -382,18 +381,43 @@ public class SingleConnectionFactoryTests {
|
|||||||
scf.setReconnectOnException(true);
|
scf.setReconnectOnException(true);
|
||||||
Field conField = ReflectionUtils.findField(SingleConnectionFactory.class, "connection");
|
Field conField = ReflectionUtils.findField(SingleConnectionFactory.class, "connection");
|
||||||
conField.setAccessible(true);
|
conField.setAccessible(true);
|
||||||
|
assertThat(scf.isRunning()).isFalse();
|
||||||
|
|
||||||
// Get connection (1st)
|
// Get connection (1st)
|
||||||
Connection con1 = scf.getConnection();
|
Connection con1 = scf.getConnection();
|
||||||
assertThat(createConnectionMethodCounter.get()).isEqualTo(1);
|
assertThat(createConnectionMethodCounter.get()).isEqualTo(1);
|
||||||
assertThat(con1).isNotNull();
|
|
||||||
assertThat(con1.getExceptionListener()).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)
|
// Get connection again, the same should be returned (shared connection till some problem)
|
||||||
Connection con2 = scf.getConnection();
|
Connection con2 = scf.getConnection();
|
||||||
assertThat(createConnectionMethodCounter.get()).isEqualTo(1);
|
assertThat(createConnectionMethodCounter.get()).isEqualTo(1);
|
||||||
assertThat(con2.getExceptionListener()).isNotNull();
|
assertThat(con2.getExceptionListener()).isNotNull();
|
||||||
assertThat(con2).isSameAs(con1);
|
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
|
// Invoke reset connection to simulate problem with connection
|
||||||
// - SCF exception listener should be invoked -> connection should be set to null
|
// - SCF exception listener should be invoked -> connection should be set to null
|
||||||
@@ -405,16 +429,21 @@ public class SingleConnectionFactoryTests {
|
|||||||
// - JMSException should be returned from FailingTestConnection
|
// - JMSException should be returned from FailingTestConnection
|
||||||
// - connection should be still null (no new connection without exception listener like before fix)
|
// - connection should be still null (no new connection without exception listener like before fix)
|
||||||
assertThatExceptionOfType(JMSException.class).isThrownBy(() -> scf.getConnection());
|
assertThatExceptionOfType(JMSException.class).isThrownBy(() -> scf.getConnection());
|
||||||
assertThat(createConnectionMethodCounter.get()).isEqualTo(2);
|
assertThat(createConnectionMethodCounter.get()).isEqualTo(4);
|
||||||
assertThat(conField.get(scf)).isNull();
|
assertThat(conField.get(scf)).isNull();
|
||||||
|
|
||||||
// Attempt to get connection again -> FailingTestConnection should be returned
|
// Attempt to get connection again -> FailingTestConnection should be returned
|
||||||
// - no JMSException is thrown, exception listener should be present
|
// - no JMSException is thrown, exception listener should be present
|
||||||
Connection con3 = scf.getConnection();
|
Connection con5 = scf.getConnection();
|
||||||
assertThat(createConnectionMethodCounter.get()).isEqualTo(3);
|
assertThat(createConnectionMethodCounter.get()).isEqualTo(5);
|
||||||
assertThat(con3).isNotNull();
|
assertThat(con5).isNotNull();
|
||||||
assertThat(con3).isSameAs(failingCon);
|
assertThat(con5).isSameAs(failingCon);
|
||||||
assertThat(con3.getExceptionListener()).isNotNull();
|
assertThat(con5.getExceptionListener()).isNotNull();
|
||||||
|
assertThat(con5).isNotSameAs(con4);
|
||||||
|
|
||||||
|
scf.destroy();
|
||||||
|
assertThat(conField.get(scf)).isNull();
|
||||||
|
assertThat(scf.isRunning()).isFalse();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
|||||||
Reference in New Issue
Block a user