diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index b133cfdb..f5c1b663 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -43,6 +43,7 @@ import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.connection.ChannelProxy; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; @@ -387,7 +388,8 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta synchronized (this.consumersMonitor) { consumersToCancel = this.consumers.stream() .filter(c -> { - boolean open = c.getChannel().isOpen() && !c.isAckFailed(); + boolean open = c.getChannel().isOpen() && !c.isAckFailed() + && !c.targetChanged(); if (open && this.messagesPerAck > 1) { try { c.ackIfNecessary(now); @@ -774,6 +776,8 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta private final long ackTimeout = DirectMessageListenerContainer.this.ackTimeout; + private final Channel targetChannel; + private int pendingAcks; private long lastAck = System.currentTimeMillis(); @@ -797,6 +801,12 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta this.connection = connection; this.queue = queue; this.ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual(); + if (channel instanceof ChannelProxy) { + this.targetChannel = ((ChannelProxy) channel).getTargetChannel(); + } + else { + this.targetChannel = null; + } } private String getQueue() { @@ -833,6 +843,15 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta return this.ackFailed; } + /** + * True if the channel is a proxy and the underlying channel has changed. + * @return true if the condition exists. + */ + boolean targetChanged() { + return this.targetChannel != null + && !((ChannelProxy) getChannel()).getTargetChannel().equals(this.targetChannel); + } + /** * Increment and return the current epoch for this consumer; consumersMonitor must * be held. diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java index f5107f39..ca10a502 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -43,6 +43,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.rabbit.connection.ChannelProxy; import org.springframework.amqp.rabbit.connection.Connection; @@ -175,7 +176,7 @@ public class DirectMessageListenerContainerMockTests { } Thread.sleep(200); consumer.get().handleDelivery("consumerTag", envelope(16), props, body); - // should get 2 acks #10 and #6 (timeout) + // should get 2 acks #10 and #16 (timeout) assertTrue(latch2.await(10, TimeUnit.SECONDS)); consumer.get().handleDelivery("consumerTag", envelope(17), props, body); verify(channel).basicAck(10L, true); @@ -311,6 +312,56 @@ public class DirectMessageListenerContainerMockTests { container.stop(); } + @Test + public void testMonitorCancelsAfterTargetChannelChanges() throws Exception { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + Connection connection = mock(Connection.class); + ChannelProxy channel = mock(ChannelProxy.class); + Channel rabbitChannel1 = mock(Channel.class); + Channel rabbitChannel2 = mock(Channel.class); + AtomicReference target = new AtomicReference<>(rabbitChannel1); + willAnswer(inv -> { + return target.get(); + }).given(channel).getTargetChannel(); + + given(connectionFactory.createConnection()).willReturn(connection); + given(connection.createChannel(anyBoolean())).willReturn(channel); + given(channel.isOpen()).willReturn(true); + given(channel.queueDeclarePassive(Mockito.anyString())) + .willAnswer(invocation -> mock(AMQP.Queue.DeclareOk.class)); + AtomicReference consumer = new AtomicReference<>(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + willAnswer(inv -> { + consumer.set(inv.getArgument(6)); + latch1.countDown(); + return "consumerTag"; + }).given(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), + anyMap(), any(Consumer.class)); + + willAnswer(inv -> { + consumer.get().handleCancelOk("consumerTag"); + latch2.countDown(); + return null; + }).given(channel).basicCancel("consumerTag"); + + DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory); + container.setQueueNames("test"); + container.setPrefetchCount(2); + container.setMonitorInterval(100); + container.setMessageListener(msg -> { + target.set(rabbitChannel2); + }); + container.setAcknowledgeMode(AcknowledgeMode.MANUAL); + container.afterPropertiesSet(); + container.start(); + + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + consumer.get().handleDelivery("consumerTag", envelope(1L), new BasicProperties(), new byte[1]); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + container.stop(); + } + private Envelope envelope(long tag) { return new Envelope(tag, false, "", ""); }