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

Fix race in test

Browse Source 
CorrelationData must have an id when correlating returns and confirms.

With no id; there is a race to deliver the confirm and return.
This commit is contained in:
Gary Russell
2020-08-28 15:38:59 -04:00
parent 3c569fb5f1
commit c014954540
1 changed files with 8 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java
View File

@@ -65,10 +65,10 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
template.setConfirmCallback((cd, a, c) -> {
confirmLatch.countDown();
if (confirmLatch.getCount() == 1) {
template.convertAndSend(QUEUE1, ((MyCD) cd).payload);
template.convertAndSend(QUEUE1, cd.getId());
}
});
template.convertAndSend("bad.exchange", "junk", "foo", new MyCD("foo"));
template.convertAndSend("bad.exchange", "junk", "foo", new CorrelationData("foo"));
assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(template.receive(QUEUE1, 10_000)).isNotNull();
}
@@ -102,11 +102,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
channel2.close();
conn.close();
assertThat(TestUtils.getPropertyValue(cf, "cachedChannelsNonTransactional", List.class).size()).isEqualTo(2);
template.convertAndSend("", QUEUE2 + "junk", "foo", new MyCD("foo"));
CorrelationData correlationData = new CorrelationData("foo");
template.convertAndSend("", QUEUE2 + "junk", "foo", correlationData);
assertThat(returnLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(cacheCount.get()).isEqualTo(1);
assertThat(returnCalledFirst.get()).isTrue();
assertThat(correlationData.getReturnedMessage()).isNotNull();
cf.destroy();
}
@@ -130,7 +132,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
channel2.close();
conn.close();
assertThat(TestUtils.getPropertyValue(cf, "cachedChannelsNonTransactional", List.class).size()).isEqualTo(2);
template.convertAndSend("", QUEUE2, "foo", new MyCD("foo"));
template.convertAndSend("", QUEUE2, "foo", new CorrelationData("foo"));
assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(cacheCount.get()).isEqualTo(1);
cf.destroy();
@@ -147,22 +149,11 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
template.setConfirmCallback((cd, a, c) -> {
confirmLatch.countDown();
});
template.convertSendAndReceive("", QUEUE3, "foo", new MyCD("foo"));
template.convertSendAndReceive("", QUEUE3, "foo", new MyCD("foo")); // listener not registered
template.convertSendAndReceive("", QUEUE3, "foo", new CorrelationData("foo"));
template.convertSendAndReceive("", QUEUE3, "foo", new CorrelationData("foo")); // listener not registered
assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(template.receive(QUEUE3, 10_000)).isNotNull();
assertThat(template.receive(QUEUE3, 10_000)).isNotNull();
}
private static class MyCD extends CorrelationData {
final String payload;
MyCD(String payload) {
this.payload = payload;
}
}
}