Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-151: Ensure Connection.close() is called on the right target connection

Browse Source
This commit is contained in:
Dave Syer
2011-04-06 09:12:35 +01:00
parent 42c45f7214
commit a7447e3ea2
2 changed files with 38 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java
View File

@@ -41,11 +41,8 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
/** Raw Rabbit Connection */
private Connection targetConnection;
/** Proxy Connection */
private Connection connection;
private SharedConnectionProxy connection;
/** Synchronization monitor for the shared Connection */
private final Object connectionMonitor = new Object();
@@ -142,14 +139,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
public final Connection createConnection() throws AmqpException {
synchronized (this.connectionMonitor) {
if (this.connection == null) {
if (this.targetConnection != null) {
RabbitUtils.closeConnection(this.targetConnection);
}
this.targetConnection = doCreateConnection();
if (logger.isInfoEnabled()) {
logger.info("Established shared Rabbit Connection: " + this.targetConnection);
}
this.connection = new SharedConnectionProxy(this.targetConnection);
this.connection = new SharedConnectionProxy(doCreateConnection());
}
this.listener.onCreate(connection);
}
@@ -164,11 +154,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
*/
public final void destroy() {
synchronized (this.connectionMonitor) {
if (this.targetConnection != null) {
listener.onClose(targetConnection);
RabbitUtils.closeConnection(this.targetConnection);
}
this.targetConnection = null;
this.connection.destroy();
this.connection = null;
}
reset();
@@ -231,9 +217,9 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
}
public Channel createChannel(boolean transactional) {
if (!target.isOpen()) {
if (target==null || !target.isOpen()) {
synchronized (this) {
if (!target.isOpen()) {
if (target==null || !target.isOpen()) {
logger.debug("Detected closed connection. Opening a new one before creating Channel.");
target = createBareConnection();
}
@@ -245,6 +231,14 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
public void close() {
}
public void destroy() {
if (this.target != null) {
listener.onClose(target);
RabbitUtils.closeConnection(this.target);
}
this.target = null;
}
public boolean isOpen() {
return target != null && target.isOpen();

25
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java
View File

@@ -5,6 +5,7 @@ import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -50,5 +51,29 @@ public class SingleConnectionFactoryTests {
verify(mockConnection, atLeastOnce()).close();
}
@Test
public void testCloseInvalidConnection() throws Exception {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class);
com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2);
// simulate a dead connection
when(mockConnection1.isOpen()).thenReturn(false);
SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
Connection connection = connectionFactory.createConnection();
// the dead connection should be discarded
connection.createChannel(false);
verify(mockConnectionFactory, times(2)).newConnection();
verify(mockConnection2, times(1)).createChannel();
connectionFactory.destroy();
verify(mockConnection2, times(1)).close();
}
}