diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java index 226a3413..8394f34e 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java @@ -176,13 +176,13 @@ public class RabbitAmqpAdmin */ @Override public void initialize() { - redeclareBeanDeclarables(); + declareDeclarableBeans(); } /** * Process bean declarables. */ - private void redeclareBeanDeclarables() { + private void declareDeclarableBeans() { if (this.applicationContext == null) { LOG.debug("no ApplicationContext has been set, cannot auto-declare Exchanges, Queues, and Bindings"); return; diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java index 479af40f..034dd85a 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java @@ -256,24 +256,25 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { } } catch (Exception ex) { - if (!handleSpecialErrors(ex, context)) { - try { - this.errorHandler.handleError(ex); - // If error handler does not re-throw an exception, treat it as a successful processing result. + try { + this.errorHandler.handleError(ex); + // If error handler does not re-throw an exception, re-check original error. + // If it is not special, treat the error handler outcome as a successful processing result. + if (!handleSpecialErrors(ex, context)) { context.accept(); } - catch (Exception rethrow) { - if (!handleSpecialErrors(rethrow, context)) { - if (this.defaultRequeue) { - context.requeue(); - } - else { - context.discard(); - } - LOG.error(rethrow, () -> - "The 'errorHandler' has thrown an exception. The '" + amqpMessage + "' is " - + (this.defaultRequeue ? "re-queued." : "discarded.")); + } + catch (Exception rethrow) { + if (!handleSpecialErrors(rethrow, context)) { + if (this.defaultRequeue) { + context.requeue(); } + else { + context.discard(); + } + LOG.error(rethrow, () -> + "The 'errorHandler' has thrown an exception. The '" + amqpMessage + "' is " + + (this.defaultRequeue ? "re-queued." : "discarded.")); } } } @@ -321,9 +322,9 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { CompletableFuture[] completableFutures = this.queueToConsumers.values().stream() .flatMap(List::stream) - .peek(Consumer::pause) .map((consumer) -> CompletableFuture.supplyAsync(() -> { + consumer.pause(); try (consumer) { while (consumer.unsettledMessageCount() > 0) { Thread.sleep(100); diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java index 98ba024e..d52f0d03 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java @@ -53,8 +53,8 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd @Override public void onAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, Consumer.@Nullable Context context) { + org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context); try { - org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context); org.springframework.messaging.Message messagingMessage = toMessagingMessage(springMessage); InvocationResult result = getHandlerAdapter() .invoke(messagingMessage, @@ -65,7 +65,7 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd } } catch (Exception ex) { - throw new ListenerExecutionFailedException("Failed to invoke listener", ex); + throw new ListenerExecutionFailedException("Failed to invoke listener", ex, springMessage); } } diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java index 8a95bb02..22b6561b 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java @@ -64,15 +64,6 @@ public class RabbitAmqpAdminTests extends RabbitAmqpTestBase { assertThat(template.receiveAndConvert("q3")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test4"); assertThat(template.receiveAndConvert("q4")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test5"); - admin.deleteQueue("q1"); - admin.deleteQueue("q2"); - admin.deleteQueue("q3"); - admin.deleteQueue("q4"); - admin.deleteExchange("e1"); - admin.deleteExchange("e2"); - admin.deleteExchange("e3"); - admin.deleteExchange("e4"); - assertThat(declarables.getDeclarablesByType(Queue.class)) .hasSize(1) .extracting(Queue::getName) @@ -92,12 +83,12 @@ public class RabbitAmqpAdminTests extends RabbitAmqpTestBase { @Bean DirectExchange e1() { - return new DirectExchange("e1", false, false); + return new DirectExchange("e1"); } @Bean Queue q1() { - return new Queue("q1", false, false, false); + return new Queue("q1"); } @Bean @@ -108,15 +99,15 @@ public class RabbitAmqpAdminTests extends RabbitAmqpTestBase { @Bean Declarables es() { return new Declarables( - new DirectExchange("e2", false, false), - new DirectExchange("e3", false, false)); + new DirectExchange("e2"), + new DirectExchange("e3")); } @Bean Declarables qs() { return new Declarables( - new Queue("q2", false, false, false), - new Queue("q3", false, false, false)); + new Queue("q2"), + new Queue("q3")); } @Bean @@ -129,8 +120,8 @@ public class RabbitAmqpAdminTests extends RabbitAmqpTestBase { @Bean Declarables ds() { return new Declarables( - new DirectExchange("e4", false, false), - new Queue("q4", false, false, false), + new DirectExchange("e4"), + new Queue("q4"), new Binding("q4", Binding.DestinationType.QUEUE, "e4", "k4", null)); } diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java index bff14523..c4dc1146 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java @@ -102,12 +102,12 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase { @Bean DirectExchange e1() { - return new DirectExchange("e1", false, false); + return new DirectExchange("e1"); } @Bean Queue q1() { - return new Queue("q1", false, false, false); + return new Queue("q1"); } @Bean diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java index 4d728412..cdd8cc03 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java @@ -16,12 +16,21 @@ package org.springframework.amqp.rabbitmq.client; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Stream; + import com.rabbitmq.client.amqp.Connection; import com.rabbitmq.client.amqp.Environment; import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder; +import org.springframework.amqp.core.Declarable; +import org.springframework.amqp.core.Declarables; +import org.springframework.amqp.core.Exchange; +import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.junit.AbstractTestContainerTests; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.Lifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.test.annotation.DirtiesContext; @@ -36,7 +45,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; */ @SpringJUnitConfig @DirtiesContext -abstract class RabbitAmqpTestBase extends AbstractTestContainerTests { +public abstract class RabbitAmqpTestBase extends AbstractTestContainerTests { @Autowired protected Environment environment; @@ -51,7 +60,16 @@ abstract class RabbitAmqpTestBase extends AbstractTestContainerTests { protected RabbitAmqpTemplate template; @Configuration - public static class AmqpCommonConfig { + public static class AmqpCommonConfig implements Lifecycle { + + @Autowired + List declarables; + + @Autowired(required = false) + List declarableContainers = new ArrayList<>(); + + @Autowired + RabbitAmqpAdmin admin; @Bean Environment environment() { @@ -77,6 +95,34 @@ abstract class RabbitAmqpTestBase extends AbstractTestContainerTests { return new RabbitAmqpTemplate(connection); } + volatile boolean running; + + @Override + public void start() { + this.running = true; + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public void stop() { + Stream.concat(this.declarables.stream(), + this.declarableContainers.stream() + .flatMap((declarables) -> declarables.getDeclarables().stream())) + .filter((declarable) -> declarable instanceof Queue || declarable instanceof Exchange) + .forEach((declarable) -> { + if (declarable instanceof Queue queue) { + this.admin.deleteQueue(queue.getName()); + } + else { + this.admin.deleteExchange(((Exchange) declarable).getName()); + } + }); + } + } } diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java new file mode 100644 index 00000000..55d83b29 --- /dev/null +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java @@ -0,0 +1,163 @@ +/* + * Copyright 2025 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbitmq.client.listener; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import com.rabbitmq.client.amqp.Connection; +import com.rabbitmq.client.amqp.Consumer; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.AmqpAcknowledgment; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.QueueBuilder; +import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.rabbit.annotation.EnableRabbit; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry; +import org.springframework.amqp.rabbitmq.client.RabbitAmqpTestBase; +import org.springframework.amqp.rabbitmq.client.config.RabbitAmqpListenerContainerFactory; +import org.springframework.amqp.support.converter.MessageConversionException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.ContextConfiguration; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + * + * @since 4.0 + */ +@ContextConfiguration +class RabbitAmqpListenerTests extends RabbitAmqpTestBase { + + @Autowired + Config config; + + @Autowired + RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry; + + @Test + @SuppressWarnings("unchecked") + void verifyAllDataIsConsumedFromQ1AndQ2() throws InterruptedException { + MessageListenerContainer testAmqpListener = + this.rabbitListenerEndpointRegistry.getListenerContainer("testAmqpListener"); + + assertThat(testAmqpListener).extracting("queueToConsumers") + .asInstanceOf(InstanceOfAssertFactories.map(String.class, List.class)) + .hasSize(2) + .values() + .flatMap(list -> (List) list) + .hasSize(4); + + List testDataList = + List.of("data1", "data2", "requeue", "data4", "data5", "discard", "data7", "data8", "discard", "data10"); + + Random random = new Random(); + + for (String testData : testDataList) { + this.template.convertAndSend((random.nextInt(2) == 0 ? "q1" : "q2"), testData); + } + + assertThat(this.config.consumeIsDone.await(10, TimeUnit.SECONDS)).isTrue(); + + assertThat(this.config.received).containsAll(testDataList); + + assertThat(this.template.receive("dlq1")).succeedsWithin(10, TimeUnit.SECONDS); + assertThat(this.template.receive("dlq1")).succeedsWithin(10, TimeUnit.SECONDS); + } + + @Configuration + @EnableRabbit + static class Config { + + @Bean + TopicExchange dlx1() { + return new TopicExchange("dlx1"); + } + + @Bean + Queue dlq1() { + return new Queue("dlq1"); + } + + @Bean + Binding dlq1Binding() { + return BindingBuilder.bind(dlq1()).to(dlx1()).with("#"); + } + + @Bean + Queue q1() { + return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build(); + } + + @Bean + Queue q2() { + return QueueBuilder.durable("q2").deadLetterExchange("dlx1").build(); + } + + @Bean(RabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME) + RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory(Connection connection) { + return new RabbitAmqpListenerContainerFactory(connection); + } + + List received = Collections.synchronizedList(new ArrayList<>()); + + CountDownLatch consumeIsDone = new CountDownLatch(10); + + @RabbitListener(queues = {"q1", "q2"}, + ackMode = "#{T(org.springframework.amqp.core.AcknowledgeMode).MANUAL}", + concurrency = "2", + id = "testAmqpListener") + void processQ1AndQ2Data(String data, AmqpAcknowledgment acknowledgment, Consumer.Context context) { + try { + if ("discard".equals(data)) { + if (!this.received.contains(data)) { + context.discard(); + } + else { + throw new MessageConversionException("Test message is rejected"); + } + } + else if ("requeue".equals(data) && !this.received.contains(data)) { + acknowledgment.acknowledge(AmqpAcknowledgment.Status.REQUEUE); + } + else { + acknowledgment.acknowledge(); + } + this.received.add(data); + } + finally { + this.consumeIsDone.countDown(); + } + } + + } + +}