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

AMQP-847: Close channel in RabbitTemplate.receive

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-847

To avoid unacked messages race condition when client timeouts, but at
this moment the message becomes available in queue, physically close
a receive channel on the `TimeoutException` from the `Future.get()`

**Cherry-pick to 2.0.x & 1.7.x**
This commit is contained in:
Artem Bilan
2018-12-11 14:18:19 -05:00
committed by Gary Russell
parent c20e216f1c
commit 41a61eb4de
2 changed files with 13 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -890,6 +890,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
}
channel.basicCancel(consumer.getConsumerTag());
if (delivery == null) {
RabbitUtils.setPhysicalCloseRequired(channel, true);
return null;
}
else {
@@ -1029,6 +1030,9 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
}
receiveMessage = buildMessageFromDelivery(delivery);
}
else {
RabbitUtils.setPhysicalCloseRequired(channel, true);
}
}
if (receiveMessage != null) {
Object receive = receiveMessage;

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

@@ -47,6 +47,7 @@ import java.io.UnsupportedEncodingException;
import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
@@ -348,6 +349,14 @@ public class RabbitTemplateIntegrationTests {
assertNull(this.template.receive(ROUTE));
}
@Test
public void testReceiveTimeoutRequeue() {
assertNull(this.template.receiveAndConvert(ROUTE, 1));
assertEquals(0,
TestUtils.getPropertyValue(this.connectionFactory, "cachedChannelsNonTransactional", List.class)
.size());
}
@Test
public void testReceiveBlockingTx() throws Exception {
this.template.convertAndSend(ROUTE, "blockTX");