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

AMQP-231 Failing Integration Tests

Browse Source 
Mock setup for connection creation in 4 tests was not
updated for AMQP-225 changes (external executor).

Listener connection retry did not check for a
closed channel.

AMQP-231 More Spurious Integration Test Failures

RabbitBindingIntegrationTests Issues

* Asserting notNull on consumer tag before it might have been returned
by the broker.

* Calling basicCancel() on the wrong channel - the rabbit 2.5.0
client sent the cancel to the broker anyhow; the newer clients
check whether the tag is for a known consumer and throw an
exception if not. The Test cases were calling basicCancel() on
the sending channel instead of the consuming channel.

MessageListenerContainerLifecycleIntegrationTests did not
wait long enough for the listener's consumer to terminate.
This commit is contained in:
Gary Russell
2012-05-12 10:42:16 -04:00
committed by Oleg Zhurakousky
parent 99fbfa16cc
commit 2dcf0ea642
4 changed files with 25 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -214,7 +214,7 @@ public class BlockingQueueConsumer {
}
passiveDeclareTries = 0;
} catch (IOException e) {
if (passiveDeclareTries > 0) {
if (passiveDeclareTries > 0 && channel.isOpen()) {
if (logger.isWarnEnabled()) {
logger.warn("Reconnect failed; retries left=" + (passiveDeclareTries-1), e);
try {

17
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
View File

@@ -63,7 +63,7 @@ public class RabbitBindingIntegrationTests {
assertEquals("message", result);
} finally {
channel.basicCancel(tag);
consumer.getChannel().basicCancel(tag);
}
return null;
@@ -102,7 +102,7 @@ public class RabbitBindingIntegrationTests {
assertEquals("message", result);
} finally {
channel.basicCancel(tag);
consumer.getChannel().basicCancel(tag);
}
return null;
@@ -239,6 +239,19 @@ public class RabbitBindingIntegrationTests {
accessor.getConnectionFactory(), new DefaultMessagePropertiesConverter(),
new ActiveObjectCounter<BlockingQueueConsumer>(), AcknowledgeMode.AUTO, true, 1, queue.getName());
consumer.start();
// wait for consumeOk...
int n = 0;
while (n++ < 100) {
if (consumer.getConsumerTag() == null) {
try {
Thread.sleep(100);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}
return consumer;
}

9
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -182,7 +183,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(new PublisherCallbackChannelImpl(mockChannel));
@@ -209,7 +210,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
PublisherCallbackChannelImpl channel1 = new PublisherCallbackChannelImpl(mockChannel);
PublisherCallbackChannelImpl channel2 = new PublisherCallbackChannelImpl(mockChannel);
@@ -278,7 +279,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(new PublisherCallbackChannelImpl(mockChannel));
@@ -317,7 +318,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
PublisherCallbackChannelImpl callbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel()).thenReturn(callbackChannel);

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java
View File

@@ -160,7 +160,10 @@ public class MessageListenerContainerLifecycleIntegrationTests {
assertEquals(concurrentConsumers, container.getActiveConsumerCount());
container.stop();
Thread.sleep(1000L);
int n = 0;
while (n++ < 100 && container.getActiveConsumerCount() > 0) {
Thread.sleep(100);
}
assertEquals(0, container.getActiveConsumerCount());
if (!transactional) {