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

GH-2994: Add tests for RabbitAmqpListenerContainer

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

* Add `@RabbitListener` tests for `RabbitAmqpMessageListener`
* Adjust `RabbitAmqpMessageListener` error handling logic to call `errorHandler` first
* Move `consumer.pause()` for `RabbitAmqpMessageListener.stop()` to the `supplyAsync()`
to initiate pause for all consumers in parallel
* Expose a failed `message` to the `ListenerExecutionFailedException` from the `RabbitAmqpMessageListenerAdapter`
* Ensure RabbitMQ objects are deleted in the end of tests
This commit is contained in:
Artem Bilan
2025-03-03 15:16:58 -05:00
parent 5be96cb920
commit 6402004702
7 changed files with 242 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java
View File

@@ -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;

33
spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java
View File

@@ -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<Void>[] 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);

4
spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java
View File

@@ -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);
}
}

25
spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java
View File

@@ -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));
}

4
spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java
View File

@@ -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

50
spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java
View File

@@ -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<Declarable> declarables;
@Autowired(required = false)
List<Declarables> 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());
}
});
}
}
}

163
spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java Normal file
View File

@@ -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<com.rabbitmq.client.amqp.Consumer>) list)
.hasSize(4);
List<String> 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<String> 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();
}
}
}
}