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

GH-935: Handle all exceptions in handleDelivery

Browse Source 
Fixes https://github.com/spring-projects/spring-amqp/issues/935

- Don't call basicCancel if already canceled
- Catch all `Exception`s
This commit is contained in:
Gary Russell
2019-03-19 13:39:49 -04:00
committed by Artem Bilan
parent e65135d134
commit 9839336cd9
2 changed files with 36 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -833,6 +833,8 @@ public class BlockingQueueConsumer {
private final class InternalConsumer extends DefaultConsumer {
boolean canceled;
InternalConsumer(Channel channel) {
super(channel);
}
@@ -880,11 +882,12 @@ public class BlockingQueueConsumer {
+ BlockingQueueConsumer.this.consumerTags.get(consumerTag)
+ "); " + BlockingQueueConsumer.this);
}
this.canceled = true;
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) {
if (logger.isDebugEnabled()) {
logger.debug("Storing delivery for consumerTag: '"
+ consumerTag + "' with deliveryTag: '" + envelope.getDeliveryTag() + "' in "
@@ -897,12 +900,13 @@ public class BlockingQueueConsumer {
RabbitUtils.setPhysicalCloseRequired(getChannel(), true);
// Defensive - should never happen
BlockingQueueConsumer.this.queue.clear();
getChannel().basicNack(envelope.getDeliveryTag(), true, true);
getChannel().basicCancel(consumerTag);
if (!this.canceled) {
getChannel().basicCancel(consumerTag);
}
try {
getChannel().close();
}
catch (TimeoutException e) {
catch (@SuppressWarnings("unused") TimeoutException e) {
// no-op
}
}
@@ -911,9 +915,12 @@ public class BlockingQueueConsumer {
BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body));
}
}
catch (InterruptedException e) {
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
catch (Exception e) {
BlockingQueueConsumer.logger.warn("Unexpected exception during delivery", e);
}
}
}
@@ -935,6 +942,7 @@ public class BlockingQueueConsumer {
}
@Override
public void handleConsumeOk(String consumerTag) {
this.consumerTag = consumerTag;
this.delegate.handleConsumeOk(consumerTag);
@@ -943,24 +951,29 @@ public class BlockingQueueConsumer {
}
}
@Override
public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
this.delegate.handleShutdownSignal(consumerTag, sig);
}
@Override
public void handleCancel(String consumerTag) throws IOException {
this.delegate.handleCancel(consumerTag);
}
@Override
public void handleCancelOk(String consumerTag) {
this.delegate.handleCancelOk(consumerTag);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
this.delegate.handleDelivery(consumerTag, envelope, properties, body);
}
@Override
public void handleRecoverOk(String consumerTag) {
this.delegate.handleRecoverOk(consumerTag);
}

24
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java
View File

@@ -27,6 +27,7 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willThrow;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -40,8 +41,10 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.logging.log4j.Level;
import org.junit.Rule;
@@ -297,7 +300,7 @@ public class BlockingQueueConsumerTests {
}
@Test
public void testDrainAndReject() throws IOException {
public void testDrainAndReject() throws IOException, TimeoutException {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
Connection connection = mock(Connection.class);
ChannelProxy channel = mock(ChannelProxy.class);
@@ -310,9 +313,18 @@ public class BlockingQueueConsumerTests {
doReturn(isOpen.get()).when(channel).isOpen();
when(channel.queueDeclarePassive(anyString()))
.then(invocation -> mock(AMQP.Queue.DeclareOk.class));
when(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
anyMap(), any(Consumer.class))).thenReturn("consumerTag");
AtomicReference<Consumer> theConsumer = new AtomicReference<>();
doAnswer(inv -> {
Consumer consumer = inv.getArgument(6);
consumer.handleConsumeOk("consumerTag");
theConsumer.set(consumer);
return "consumerTag";
}).when(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
anyMap(), any(Consumer.class));
doAnswer(inv -> {
theConsumer.get().handleCancelOk("consumerTag");
return null;
}).when(channel).basicCancel("consumerTag");
BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory,
new DefaultMessagePropertiesConverter(), new ActiveObjectCounter<BlockingQueueConsumer>(),
AcknowledgeMode.AUTO, true, 2, "test");
@@ -337,9 +349,7 @@ public class BlockingQueueConsumerTests {
envelope = new Envelope(3, false, "foo", "bar");
consumer.handleDelivery("consumerTag", envelope, props, new byte[0]);
assertThat(TestUtils.getPropertyValue(blockingQueueConsumer, "queue", BlockingQueue.class).size(), equalTo(0));
verify(channel).basicNack(3, true, true);
verify(channel, times(2)).basicCancel("consumerTag");
verify(channel, times(1)).basicCancel("consumerTag");
}
}