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

SMLC: defer counter release until fetched messages are processed

Browse Source 
Currently, the `SimpleMessageListenerContainer` is stopped immediately
when `cancelOK` is received.

The expectation do not stop an application context until all the fetched
messages are processed.
Therefore, move `this.activeObjectCounter.release(this);` to the `BlockingQueueConsumer.nextMessage()`
if the internal queue is empty and `cancelled` has been requested.

* Adjust all the SMLC tests for a shorter `receiveTimeout` to not have blocking for nothing
* Also add `System.setProperty("spring.amqp.deserialization.trust.all", "true");` to be
able to run tests from IDE

**Cherry-pick to `3.0.x`**
This commit is contained in:
Artem Bilan
2024-02-05 12:49:44 -05:00
parent aa46e3cd0a
commit ca51ca62a6
35 changed files with 174 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/examples/TestRabbitTemplateTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2024 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.
@@ -135,6 +135,7 @@ public class TestRabbitTemplateTests {
public SimpleMessageListenerContainer smlc1() throws IOException {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueueNames("foo", "bar");
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(new Object() {
@SuppressWarnings("unused")

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

@@ -551,6 +551,7 @@ public class BlockingQueueConsumer {
}
Message message = handle(this.queue.poll(timeout, TimeUnit.MILLISECONDS));
if (message == null && this.cancelled.get()) {
this.activeObjectCounter.release(this);
throw new ConsumerCancelledException();
}
return message;
@@ -990,7 +991,6 @@ public class BlockingQueueConsumer {
+ "); " + BlockingQueueConsumer.this);
}
this.canceled = true;
BlockingQueueConsumer.this.activeObjectCounter.release(BlockingQueueConsumer.this);
}
@Override

16
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2024 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.
@@ -382,17 +382,17 @@ public class AsyncRabbitTemplateTests {
@Test
void ctorCoverage() {
AsyncRabbitTemplate template = new AsyncRabbitTemplate(mock(ConnectionFactory.class), "ex", "rk");
assertThat(template).extracting(t -> t.getRabbitTemplate())
assertThat(template).extracting(AsyncRabbitTemplate::getRabbitTemplate)
.extracting("exchange")
.isEqualTo("ex");
assertThat(template).extracting(t -> t.getRabbitTemplate())
assertThat(template).extracting(AsyncRabbitTemplate::getRabbitTemplate)
.extracting("routingKey")
.isEqualTo("rk");
template = new AsyncRabbitTemplate(mock(ConnectionFactory.class), "ex", "rk", "rq");
assertThat(template).extracting(t -> t.getRabbitTemplate())
assertThat(template).extracting(AsyncRabbitTemplate::getRabbitTemplate)
.extracting("exchange")
.isEqualTo("ex");
assertThat(template).extracting(t -> t.getRabbitTemplate())
assertThat(template).extracting(AsyncRabbitTemplate::getRabbitTemplate)
.extracting("routingKey")
.isEqualTo("rk");
assertThat(template)
@@ -402,10 +402,10 @@ public class AsyncRabbitTemplateTests {
.extracting("queueNames")
.isEqualTo(new String[] { "rq" });
template = new AsyncRabbitTemplate(mock(ConnectionFactory.class), "ex", "rk", "rq", "ra");
assertThat(template).extracting(t -> t.getRabbitTemplate())
assertThat(template).extracting(AsyncRabbitTemplate::getRabbitTemplate)
.extracting("exchange")
.isEqualTo("ex");
assertThat(template).extracting(t -> t.getRabbitTemplate())
assertThat(template).extracting(AsyncRabbitTemplate::getRabbitTemplate)
.extracting("routingKey")
.isEqualTo("rk");
assertThat(template)
@@ -522,6 +522,7 @@ public class AsyncRabbitTemplateTests {
@Primary
public SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setReceiveTimeout(10);
container.setAfterReceivePostProcessors(new GUnzipPostProcessor());
container.setQueueNames(replies().getName());
return container;
@@ -540,6 +541,7 @@ public class AsyncRabbitTemplateTests {
@Bean
public SimpleMessageListenerContainer remoteContainer(ConnectionFactory connectionFactory) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setReceiveTimeout(10);
container.setQueueNames(requests().getName());
container.setAfterReceivePostProcessors(new GUnzipPostProcessor());
MessageListenerAdapter messageListener =

20
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2020 the original author or authors.
* Copyright 2014-2024 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.
@@ -142,6 +142,7 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
// Resolve the container and invoke a message on it
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setReceiveTimeout(10);
endpoint.setupListenerContainer(container);
MessagingMessageListenerAdapter listener = (MessagingMessageListenerAdapter) container.getMessageListener();
@@ -158,9 +159,9 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
}
/**
* Test for {@link CustomBean} and an manually endpoint registered
* Test for {@link CustomBean} and a manually registered endpoint
* with "myCustomEndpointId". The custom endpoint does not provide
* any factory so it's registered with the default one
* any factory, so it's registered with the default one
*/
public void testCustomConfiguration(ApplicationContext context) {
RabbitListenerContainerTestFactory defaultFactory =
@@ -171,14 +172,15 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
assertThat(customFactory.getListenerContainers()).hasSize(1);
RabbitListenerEndpoint endpoint = defaultFactory.getListenerContainers().get(0).getEndpoint();
assertThat(endpoint.getClass()).as("Wrong endpoint type").isEqualTo(SimpleRabbitListenerEndpoint.class);
assertThat(((SimpleRabbitListenerEndpoint) endpoint).getMessageListener()).as("Wrong listener set in custom endpoint").isEqualTo(context.getBean("simpleMessageListener"));
assertThat(((SimpleRabbitListenerEndpoint) endpoint).getMessageListener())
.as("Wrong listener set in custom endpoint").isEqualTo(context.getBean("simpleMessageListener"));
RabbitListenerEndpointRegistry customRegistry =
context.getBean("customRegistry", RabbitListenerEndpointRegistry.class);
assertThat(customRegistry.getListenerContainerIds().size()).as("Wrong number of containers in the registry").isEqualTo(2);
assertThat(customRegistry.getListenerContainers().size()).as("Wrong number of containers in the registry").isEqualTo(2);
assertThat(customRegistry.getListenerContainer("listenerId")).as("Container with custom id on the annotation should be found").isNotNull();
assertThat(customRegistry.getListenerContainer("myCustomEndpointId")).as("Container created with custom id should be found").isNotNull();
assertThat(customRegistry.getListenerContainerIds()).hasSize(2);
assertThat(customRegistry.getListenerContainers()).hasSize(2);
assertThat(customRegistry.getListenerContainer("listenerId")).isNotNull();
assertThat(customRegistry.getListenerContainer("myCustomEndpointId")).isNotNull();
}
/**
@@ -205,7 +207,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
/**
* Test for {@link ValidationBean} with a validator ({@link TestValidator}) specified
* in a custom {@link org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory}.
*
* The test should throw a {@link org.springframework.amqp.rabbit.support.ListenerExecutionFailedException}
*/
public void testRabbitHandlerMethodFactoryConfiguration(ApplicationContext context) throws Exception {
@@ -216,6 +217,7 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
simpleFactory.getListenerContainers().get(0).getEndpoint();
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setReceiveTimeout(10);
endpoint.setupListenerContainer(container);
MessagingMessageListenerAdapter listener = (MessagingMessageListenerAdapter) container.getMessageListener();

13
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2023 the original author or authors.
* Copyright 2014-2024 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.
@@ -41,6 +41,7 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -248,6 +249,7 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
public static void setUp() {
System.setProperty(RabbitListenerAnnotationBeanPostProcessor.RABBIT_EMPTY_STRING_ARGUMENTS_PROPERTY,
"test-empty");
System.setProperty("spring.amqp.deserialization.trust.all", "true");
RabbitAvailableCondition.getBrokerRunning().removeExchanges("auto.exch.tx",
"auto.exch",
"auto.exch.fanout",
@@ -826,7 +828,7 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
}
@Test
public void testHeadersExchange() throws Exception {
public void testHeadersExchange() {
assertThat(rabbitTemplate.convertSendAndReceive("auto.headers", "", "foo",
message -> {
message.getMessageProperties().getHeaders().put("foo", "bar");
@@ -845,7 +847,7 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
this.rabbitTemplate.convertAndSend("amqp656", "foo");
assertThat(this.rabbitTemplate.receiveAndConvert("amqp656dlq", 10000)).isEqualTo("foo");
try {
Map<String, Object> amqp656 = await().until(() -> queueInfo("amqp656"), q -> q != null);
Map<String, Object> amqp656 = await().until(() -> queueInfo("amqp656"), Objects::nonNull);
if (amqp656 != null) {
assertThat(arguments(amqp656).get("test-empty")).isEqualTo("");
assertThat(arguments(amqp656).get("test-null")).isEqualTo("undefined");
@@ -960,7 +962,7 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
catch (@SuppressWarnings("unused") Exception e) {
return null;
}
}, tim -> tim != null);
}, Objects::nonNull);
assertThat(timer.count()).isEqualTo(1L);
}
@@ -1786,6 +1788,7 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
factory.setBatchListener(true);
factory.setBatchSize(2);
factory.setConsumerBatchEnabled(true);
factory.setReceiveTimeout(10L);
return factory;
}
@@ -1861,7 +1864,7 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
@Bean
public AtomicReference<Throwable> errorHandlerError() {
return new AtomicReference<Throwable>();
return new AtomicReference<>();
}
@Bean

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MockMultiRabbitTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 the original author or authors.
* Copyright 2020-2024 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.
@@ -83,6 +83,7 @@ class MockMultiRabbitTests {
Assertions.assertThat(methodEndpoint.getMethod()).isNotNull();
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
listenerContainer.setReceiveTimeout(10);
methodEndpoint.setupListenerContainer(listenerContainer);
Assertions.assertThat(listenerContainer.getMessageListener()).isNotNull();
});

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2024 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.
@@ -86,6 +86,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
assertThat(methodEndpoint.getMethod()).isNotNull();
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
listenerContainer.setReceiveTimeout(10);
methodEndpoint.setupListenerContainer(listenerContainer);
assertThat(listenerContainer.getMessageListener()).isNotNull();
@@ -114,6 +115,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
assertThat(iterator.next()).isEqualTo("secondQueue");
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
listenerContainer.setReceiveTimeout(10);
methodEndpoint.setupListenerContainer(listenerContainer);
assertThat(listenerContainer.getMessageListener()).isNotNull();

11
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2024 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.
@@ -63,11 +63,11 @@ import reactor.core.publisher.Mono;
*/
public class LocalizedQueueConnectionFactoryTests {
private final Map<String, Channel> channels = new HashMap<String, Channel>();
private final Map<String, Channel> channels = new HashMap<>();
private final Map<String, Consumer> consumers = new HashMap<String, Consumer>();
private final Map<String, Consumer> consumers = new HashMap<>();
private final Map<String, String> consumerTags = new HashMap<String, String>();
private final Map<String, String> consumerTags = new HashMap<>();
@Test
public void testFailOver() throws Exception {
@@ -83,7 +83,7 @@ public class LocalizedQueueConnectionFactoryTests {
final AtomicBoolean firstServer = new AtomicBoolean(true);
final WebClient client1 = doCreateClient(adminUris[0], username, password, nodes[0]);
final WebClient client2 = doCreateClient(adminUris[1], username, password, nodes[1]);
final Map<String, ConnectionFactory> mockCFs = new HashMap<String, ConnectionFactory>();
final Map<String, ConnectionFactory> mockCFs = new HashMap<>();
CountDownLatch latch1 = new CountDownLatch(1);
CountDownLatch latch2 = new CountDownLatch(1);
mockCFs.put(rabbit1, mockCF(rabbit1, latch1));
@@ -116,6 +116,7 @@ public class LocalizedQueueConnectionFactoryTests {
willAnswer(new CallsRealMethods()).given(logger).debug(anyString());
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(lqcf);
container.setReceiveTimeout(10);
container.setQueueNames("q");
container.afterPropertiesSet();
container.start();

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2024 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.
@@ -167,7 +167,7 @@ public class RoutingConnectionFactoryTests {
public void testAbstractRoutingConnectionFactoryWithListenerContainer() {
ConnectionFactory connectionFactory1 = mock(ConnectionFactory.class);
ConnectionFactory connectionFactory2 = mock(ConnectionFactory.class);
Map<Object, ConnectionFactory> factories = new HashMap<Object, ConnectionFactory>(2);
Map<Object, ConnectionFactory> factories = new HashMap<>(2);
factories.put("[baz]", connectionFactory1);
factories.put("[foo,bar]", connectionFactory2);
ConnectionFactory defaultConnectionFactory = mock(ConnectionFactory.class);
@@ -178,6 +178,7 @@ public class RoutingConnectionFactoryTests {
connectionFactory.setTargetConnectionFactories(factories);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setReceiveTimeout(10);
container.setQueueNames("foo", "bar");
container.afterPropertiesSet();
container.start();

22
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-2024 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.
@@ -236,14 +236,14 @@ public class BatchingRabbitTemplateTests {
@Test
void testDebatchSMLCSplit() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setReceiveTimeout(100);
container.setReceiveTimeout(10);
testDebatchByContainer(container, false);
}
@Test
void testDebatchSMLC() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setReceiveTimeout(100);
container.setReceiveTimeout(10);
testDebatchByContainer(container, true);
}
@@ -303,16 +303,16 @@ public class BatchingRabbitTemplateTests {
@Test
public void testDebatchByContainerPerformance() throws Exception {
final List<Message> received = new ArrayList<Message>();
final List<Message> received = new ArrayList<>();
int count = 100000;
final CountDownLatch latch = new CountDownLatch(count);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setQueueNames(ROUTE);
container.setMessageListener((MessageListener) message -> {
container.setMessageListener(message -> {
received.add(message);
latch.countDown();
});
container.setReceiveTimeout(100);
container.setReceiveTimeout(10);
container.setPrefetchCount(1000);
container.setBatchSize(1000);
container.afterPropertiesSet();
@@ -344,8 +344,8 @@ public class BatchingRabbitTemplateTests {
public void testDebatchByContainerBadMessageRejected() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setQueueNames(ROUTE);
container.setMessageListener((MessageListener) message -> { });
container.setReceiveTimeout(100);
container.setMessageListener(message -> { });
container.setReceiveTimeout(10);
ConditionalRejectingErrorHandler errorHandler = new ConditionalRejectingErrorHandler();
container.setErrorHandler(errorHandler);
container.afterPropertiesSet();
@@ -632,15 +632,15 @@ public class BatchingRabbitTemplateTests {
@Test
public void testCompressionWithContainer() throws Exception {
final List<Message> received = new ArrayList<Message>();
final List<Message> received = new ArrayList<>();
final CountDownLatch latch = new CountDownLatch(2);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setQueueNames(ROUTE);
container.setMessageListener((MessageListener) message -> {
container.setMessageListener(message -> {
received.add(message);
latch.countDown();
});
container.setReceiveTimeout(100);
container.setReceiveTimeout(10);
container.setAfterReceivePostProcessors(new DelegatingDecompressingPostProcessor());
container.afterPropertiesSet();
container.start();

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-2024 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.
@@ -176,6 +176,7 @@ public class FixedReplyQueueDeadLetterTests extends NeedsManagementTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueues(replyQueue());
container.setReceiveTimeout(10);
container.setMessageListener(fixedReplyQRabbitTemplate());
return container;
}
@@ -188,6 +189,7 @@ public class FixedReplyQueueDeadLetterTests extends NeedsManagementTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueues(requestQueue());
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(new PojoListener()));
return container;
}
@@ -200,6 +202,7 @@ public class FixedReplyQueueDeadLetterTests extends NeedsManagementTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueues(dlq());
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(deadListener()));
return container;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2024 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.
@@ -722,6 +722,7 @@ public class RabbitTemplateIntegrationTests {
template.setReplyTimeout(10000);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory);
container.setQueues(replyQueue);
container.setReceiveTimeout(10);
container.setMessageListener(template);
container.afterPropertiesSet();
container.start();
@@ -729,10 +730,10 @@ public class RabbitTemplateIntegrationTests {
messageProperties.setCorrelationId("myCorrelationId");
Message message = new Message("test-message".getBytes(), messageProperties);
Message reply = template.sendAndReceive(message);
assertThat(new String(received.get(1000, TimeUnit.MILLISECONDS).getBody())).isEqualTo(new String(message.getBody()));
assertThat(received.get(1000, TimeUnit.MILLISECONDS).getBody()).isEqualTo(message.getBody());
assertThat(reply).as("Reply is expected").isNotNull();
assertThat(remoteCorrelationId.get()).isEqualTo("myCorrelationId");
assertThat(new String(reply.getBody())).isEqualTo(new String(message.getBody()));
assertThat(reply.getBody()).isEqualTo(message.getBody());
// Message was consumed so nothing left on queue
reply = template.receive();
assertThat(reply).isEqualTo(null);
@@ -1236,12 +1237,13 @@ public class RabbitTemplateIntegrationTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(template.getConnectionFactory());
container.setQueues(REPLY_QUEUE);
container.setReceiveTimeout(10);
container.setMessageListener(template);
container.start();
int count = 10;
final Map<Double, Object> results = new ConcurrentHashMap<Double, Object>();
final Map<Double, Object> results = new ConcurrentHashMap<>();
ExecutorService executor = Executors.newFixedThreadPool(10);
@@ -1344,7 +1346,8 @@ public class RabbitTemplateIntegrationTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(template.getConnectionFactory());
container.setQueueNames(ROUTE);
final AtomicReference<String> replyToWas = new AtomicReference<String>();
container.setReceiveTimeout(10);
final AtomicReference<String> replyToWas = new AtomicReference<>();
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new Object() {
@SuppressWarnings("unused")
@@ -1394,7 +1397,7 @@ public class RabbitTemplateIntegrationTests {
});
messageListener.setBeforeSendReplyPostProcessors(new GZipPostProcessor());
container.setMessageListener(messageListener);
container.setReceiveTimeout(100);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
RabbitTemplate template = createSendAndReceiveRabbitTemplate(this.template.getConnectionFactory());

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2021 the original author or authors.
* Copyright 2017-2024 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.
@@ -93,6 +93,7 @@ public class RabbitTemplateMPPIntegrationTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.config.cf());
try {
container.setQueueNames(REPLIES);
container.setReceiveTimeout(10);
container.setMessageListener(this.template);
container.afterPropertiesSet();
container.start();

30
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2024 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.
@@ -29,6 +29,7 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.ConcurrentModificationException;
@@ -219,13 +220,14 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Test
public void testPublisherConfirmWithSendAndReceive() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<CorrelationData> confirmCD = new AtomicReference<CorrelationData>();
final AtomicReference<CorrelationData> confirmCD = new AtomicReference<>();
templateWithConfirmsEnabled.setConfirmCallback((correlationData, ack, cause) -> {
confirmCD.set(correlationData);
latch.countDown();
});
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactoryWithConfirmsEnabled);
container.setQueueNames(ROUTE);
container.setReceiveTimeout(10);
container.setMessageListener(
new MessageListenerAdapter((ReplyingMessageListener<String, String>) String::toUpperCase));
container.start();
@@ -285,7 +287,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Test
public void testPublisherReturns() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final List<Message> returns = new ArrayList<Message>();
final List<Message> returns = new ArrayList<>();
templateWithReturnsEnabled.setReturnsCallback((returned) -> {
returns.add(returned.getMessage());
latch.countDown();
@@ -295,13 +297,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
assertThat(latch.await(1000, TimeUnit.MILLISECONDS)).isTrue();
assertThat(returns).hasSize(1);
Message message = returns.get(0);
assertThat(new String(message.getBody(), "utf-8")).isEqualTo("message");
assertThat(new String(message.getBody(), StandardCharsets.UTF_8)).isEqualTo("message");
}
@Test
public void testPublisherReturnsWithMandatoryExpression() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final List<Message> returns = new ArrayList<Message>();
final List<Message> returns = new ArrayList<>();
templateWithReturnsEnabled.setReturnsCallback((returned) -> {
returns.add(returned.getMessage());
latch.countDown();
@@ -313,7 +315,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
assertThat(latch.await(1000, TimeUnit.MILLISECONDS)).isTrue();
assertThat(returns).hasSize(1);
Message message = returns.get(0);
assertThat(new String(message.getBody(), "utf-8")).isEqualTo("message");
assertThat(new String(message.getBody(), StandardCharsets.UTF_8)).isEqualTo("message");
}
@Test
@@ -416,7 +418,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
exec.shutdown();
assertThat(exec.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
ccf.destroy();
await().until(() -> pendingConfirms.size() == 0);
await().until(pendingConfirms::isEmpty);
}
@Test
@@ -500,7 +502,6 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
/**
* Tests that piggy-backed confirms (multiple=true) are distributed to the proper
* template.
* @throws Exception
*/
@Test
public void testPublisherConfirmMultipleWithTwoListeners() throws Exception {
@@ -523,7 +524,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
final RabbitTemplate template1 = new RabbitTemplate(ccf);
final Set<String> confirms = new HashSet<String>();
final Set<String> confirms = new HashSet<>();
final CountDownLatch latch1 = new CountDownLatch(1);
template1.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
@@ -638,8 +639,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Test
public void testNackForBadExchange() throws Exception {
final AtomicBoolean nack = new AtomicBoolean(true);
final AtomicReference<CorrelationData> correlation = new AtomicReference<CorrelationData>();
final AtomicReference<String> reason = new AtomicReference<String>();
final AtomicReference<CorrelationData> correlation = new AtomicReference<>();
final AtomicReference<String> reason = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(2);
this.templateWithConfirmsEnabled.setConfirmCallback((correlationData, ack, cause) -> {
nack.set(ack);
@@ -648,7 +649,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
latch.countDown();
});
Log logger = spy(TestUtils.getPropertyValue(connectionFactoryWithConfirmsEnabled, "logger", Log.class));
final AtomicReference<String> log = new AtomicReference<String>();
final AtomicReference<String> log = new AtomicReference<>();
willAnswer(invocation -> {
log.set((String) invocation.getArguments()[0]);
invocation.callRealMethod();
@@ -735,7 +736,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
* where the close can be detected. Run the test to verify these (and any future calls
* that are added) properly emit the nacks.
*
* The following will detect proper operation if any more calls are added in future.
* The following will detect proper operation if any more calls are added in the future.
*/
for (int i = 100; i < 110; i++) {
testPublisherConfirmCloseConcurrency(i);
@@ -792,7 +793,6 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
exec.shutdownNow();
}
@SuppressWarnings("unchecked")
@Test
public void testPublisherCallbackChannelImplCloseWithPending() throws Exception {
@@ -825,7 +825,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
Map<?, ?> pending = TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class);
await().until(() -> pending.size() == 0);
await().until(pending::isEmpty);
}
@Test

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BrokerDeclaredQueueNameTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2023 the original author or authors.
* Copyright 2018-2024 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.
@@ -171,7 +171,7 @@ public class BrokerDeclaredQueueNameTests {
@Bean
public AbstractMessageListenerContainer container() {
AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(cf());
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf());
container.setQueues(queue1());
container.setMessageListener(m -> {
message().set(m);
@@ -181,6 +181,7 @@ public class BrokerDeclaredQueueNameTests {
container.setFailedDeclarationRetryInterval(100);
container.setMissingQueuesFatal(false);
container.setRecoveryInterval(100);
container.setReceiveTimeout(10);
container.setAutoStartup(false);
return container;
}

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerAdminTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021 the original author or authors.
* Copyright 2021-2024 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.
@@ -45,6 +45,7 @@ public class ContainerAdminTests {
parent.refresh();
GenericApplicationContext child = new GenericApplicationContext(parent);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
container.setReceiveTimeout(10);
child.registerBean(SimpleMessageListenerContainer.class, () -> container);
child.refresh();
container.start();

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2024 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.
@@ -144,6 +144,7 @@ public class ContainerInitializationTests {
public SimpleMessageListenerContainer container() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueueNames(TEST_MISMATCH);
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(new Object() {
@SuppressWarnings("unused")

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java
View File

@@ -46,6 +46,7 @@ public class ContainerShutDownTests {
@Test
public void testUninterruptibleListenerSMLC() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setReceiveTimeout(10);
testUninterruptibleListener(container);
}
@@ -101,6 +102,7 @@ public class ContainerShutDownTests {
@Test
public void consumersCorrectlyCancelledOnShutdownSMLC() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setReceiveTimeout(10);
consumersCorrectlyCancelledOnShutdown(container);
}

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2024 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.
@@ -170,6 +170,7 @@ public class JavaConfigFixedReplyQueueTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueues(replyQueue());
container.setReceiveTimeout(10);
container.setMessageListener(fixedReplyQRabbitTemplate());
return container;
}
@@ -182,6 +183,7 @@ public class JavaConfigFixedReplyQueueTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueues(requestQueue());
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(new PojoListener()));
return container;
}
@@ -194,6 +196,7 @@ public class JavaConfigFixedReplyQueueTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueues(replyQueue());
container.setReceiveTimeout(10);
container.setMessageListener(fixedReplyQRabbitTemplateWrongQueue());
container.setAutoStartup(false);
return container;

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedSMLCTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2024 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.
@@ -27,7 +27,8 @@ public class LocallyTransactedSMLCTests extends LocallyTransactedTests {
@Override
protected AbstractMessageListenerContainer createContainer(AbstractConnectionFactory connectionFactory) {
AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setReceiveTimeout(10);
return container;
}

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2024 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.
@@ -101,6 +101,7 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
RabbitTemplate template = this.createTemplate(1);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory());
container.setQueues(QUEUE);
container.setReceiveTimeout(10);
final CountDownLatch messageReceived = new CountDownLatch(1);
final CountDownLatch spiedQLogger = new CountDownLatch(1);
final CountDownLatch errorHandled = new CountDownLatch(1);
@@ -108,7 +109,7 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
errorHandled.countDown();
throw new AmqpRejectAndDontRequeueException("foo", t);
});
container.setMessageListener((MessageListener) message -> {
container.setMessageListener(message -> {
try {
messageReceived.countDown();
spiedQLogger.await(10, TimeUnit.SECONDS);

23
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2024 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.
@@ -38,7 +38,6 @@ import org.mockito.Mockito;
import org.springframework.amqp.AmqpIllegalStateException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -71,9 +70,9 @@ import com.rabbitmq.client.DnsRecordIpAddressResolver;
*/
@RabbitAvailable(queues = MessageListenerContainerLifecycleIntegrationTests.TEST_QUEUE)
@LongRunning
@LogLevels(classes = { RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
MessageListenerContainerLifecycleIntegrationTests.class }, level = "INFO")
@LogLevels(classes = {RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
MessageListenerContainerLifecycleIntegrationTests.class}, level = "INFO")
public class MessageListenerContainerLifecycleIntegrationTests {
public static final String TEST_QUEUE = "test.queue.MessageListenerContainerLifecycleIntegrationTests";
@@ -84,6 +83,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
private enum TransactionMode {
ON, OFF, PREFETCH, PREFETCH_NO_TX;
public boolean isTransactional() {
return this != OFF && this != PREFETCH_NO_TX;
}
@@ -103,6 +103,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
private enum Concurrency {
LOW(1), HIGH(5);
private final int value;
Concurrency(int value) {
@@ -116,6 +117,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
private enum MessageCount {
LOW(1), MEDIUM(20), HIGH(500);
private final int value;
MessageCount(int value) {
@@ -199,7 +201,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
cf.setUsername("foo");
final CachingConnectionFactory connectionFactory = new CachingConnectionFactory(cf);
assertThatExceptionOfType(AmqpIllegalStateException.class).isThrownBy(() ->
doTest(MessageCount.LOW, Concurrency.LOW, TransactionMode.OFF, template, connectionFactory))
doTest(MessageCount.LOW, Concurrency.LOW, TransactionMode.OFF, template, connectionFactory))
.withCauseExactlyInstanceOf(FatalListenerStartupException.class);
((DisposableBean) template.getConnectionFactory()).destroy();
}
@@ -334,7 +336,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
final AtomicInteger received = new AtomicInteger();
final CountDownLatch awaitConsumeFirst = new CountDownLatch(5);
final CountDownLatch awaitConsumeSecond = new CountDownLatch(10);
container.setMessageListener((MessageListener) message -> {
container.setMessageListener(message -> {
try {
awaitStart1.countDown();
prefetched.await(10, TimeUnit.SECONDS);
@@ -353,6 +355,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
container.setPrefetchCount(5);
container.setQueueNames(queue.getName());
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
@@ -364,7 +367,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
.getPropertyValue(container, "consumers");
await().until(() -> {
if (consumers.size() > 0
&& TestUtils.getPropertyValue(consumers.iterator().next(), "queue", BlockingQueue.class).size() > 3) {
&& TestUtils.getPropertyValue(consumers.iterator().next(), "queue", BlockingQueue.class).size() > 3) {
prefetched.countDown();
return true;
}
@@ -411,6 +414,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
DirectFieldAccessor dfa = new DirectFieldAccessor(container);
dfa.setPropertyValue("logger", log);
container.setQueues(queue);
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter());
container.afterPropertiesSet();
container.start();
@@ -484,7 +488,8 @@ public class MessageListenerContainerLifecycleIntegrationTests {
public SimpleMessageListenerContainer container() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueues(queue);
container.setMessageListener((MessageListener) message -> {
container.setReceiveTimeout(10);
container.setMessageListener(message -> {
try {
consumerLatch().countDown();
Thread.sleep(500);

7
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2024 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.
@@ -90,8 +90,8 @@ public class MessageListenerContainerMultipleQueueIntegrationTests {
messageConverter.setCreateMessageIds(true);
template.setMessageConverter(messageConverter);
for (int i = 0; i < messageCount; i++) {
template.convertAndSend(queue1.getName(), Integer.valueOf(i));
template.convertAndSend(queue2.getName(), Integer.valueOf(i));
template.convertAndSend(queue1.getName(), i);
template.convertAndSend(queue2.getName(), i);
}
final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
final CountDownLatch latch = new CountDownLatch(messageCount * 2);
@@ -100,6 +100,7 @@ public class MessageListenerContainerMultipleQueueIntegrationTests {
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setChannelTransacted(true);
container.setConcurrentConsumers(concurrentConsumers);
container.setReceiveTimeout(10);
configurer.configure(container);
container.afterPropertiesSet();
container.start();

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2021 the original author or authors.
* Copyright 2002-2024 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.
@@ -116,7 +116,7 @@ public class MessageListenerContainerRetryIntegrationTests {
container.setBatchSize(2);
final CountDownLatch latch = new CountDownLatch(1);
container.setAdviceChain(new Advice[] { createRetryInterceptor(latch, stateful, true) });
container.setAdviceChain(createRetryInterceptor(latch, stateful, true));
container.setQueueNames(queue.getName());
container.setReceiveTimeout(50);
@@ -252,7 +252,7 @@ public class MessageListenerContainerRetryIntegrationTests {
container.setConcurrentConsumers(concurrentConsumers);
final CountDownLatch latch = new CountDownLatch(failedMessageCount);
container.setAdviceChain(new Advice[] { createRetryInterceptor(latch, stateful) });
container.setAdviceChain(createRetryInterceptor(latch, stateful));
container.setQueueNames(queue.getName());
container.setReceiveTimeout(50);

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021 the original author or authors.
* Copyright 2021-2024 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.
@@ -88,6 +88,7 @@ public class MessageListenerContainerTxSynchTests {
mlc.setQueueNames("foo");
mlc.setTaskExecutor(exec);
mlc.setChannelTransacted(true);
mlc.setReceiveTimeout(10);
CountDownLatch latch2 = new CountDownLatch(1);
mlc.setMessageListener(msg -> {
template.convertAndSend("foo", "bar");

1
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java
View File

@@ -130,6 +130,7 @@ public class MessageListenerManualAckIntegrationTests {
container.setConcurrentConsumers(concurrentConsumers);
container.setChannelTransacted(transactional);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
return container;

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2024 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.
@@ -132,6 +132,7 @@ public class MessageListenerTxSizeIntegrationTests {
container.setConcurrentConsumers(concurrentConsumers);
container.setChannelTransacted(transactional);
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
return container;
@@ -139,7 +140,7 @@ public class MessageListenerTxSizeIntegrationTests {
public class TestListener implements ChannelAwareMessageListener {
private final ThreadLocal<Integer> count = new ThreadLocal<Integer>();
private final ThreadLocal<Integer> count = new ThreadLocal<>();
private final CountDownLatch latch;
@@ -174,6 +175,7 @@ public class MessageListenerTxSizeIntegrationTests {
latch.countDown();
}
}
}
}

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/QueueDeclarationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023 the original author or authors.
* Copyright 2023-2024 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.
@@ -116,6 +116,7 @@ public class QueueDeclarationTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setQueueNames("test");
container.setPrefetchCount(2);
container.setReceiveTimeout(10);
container.setAmqpAdmin(admin);
container.afterPropertiesSet();
return container;

8
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2024 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.
@@ -320,6 +320,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch)));
container.setQueueNames(queue.getName());
container.setConcurrentConsumers(2);
container.setReceiveTimeout(10);
GenericApplicationContext context = new GenericApplicationContext();
context.getBeanFactory().registerSingleton("foo", queue);
context.refresh();
@@ -354,6 +355,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
new SimpleMessageListenerContainer(template.getConnectionFactory());
container1.setMessageListener(new MessageListenerAdapter(new PojoListener(latch1)));
container1.setQueueNames(queue.getName());
container1.setReceiveTimeout(10);
GenericApplicationContext context = new GenericApplicationContext();
context.getBeanFactory().registerSingleton("foo", queue);
context.refresh();
@@ -375,6 +377,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
container2.setQueueNames(queue.getName());
container2.setApplicationContext(context);
container2.setRecoveryInterval(1000);
container2.setReceiveTimeout(10);
container2.setExclusive(true); // not really necessary, but likely people will make all consumers exclusive.
final AtomicReference<ListenerContainerConsumerFailedEvent> eventRef = new AtomicReference<>();
final CountDownLatch consumeLatch2 = new CountDownLatch(1);
@@ -459,6 +462,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
container.setQueueNames(queue.getName());
container.setRecoveryInterval(500);
container.setGlobalQos(true);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
@@ -510,6 +514,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch)));
container.setQueueNames(queue.getName());
container.setRecoveryInterval(500);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
@@ -537,6 +542,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
container.setDeclarationRetries(1);
container.setFailedDeclarationRetryInterval(100);
container.setRetryDeclarationInterval(30000);
container.setReceiveTimeout(10);
container.setApplicationEventPublisher(event -> {
if (event instanceof MissingQueueEvent) {
missingLatch.countDown();

12
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2024 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.
@@ -26,7 +26,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
@@ -90,6 +89,7 @@ public class SimpleMessageListenerContainerLongTests {
container.setAutoStartup(false);
container.setConcurrentConsumers(2);
container.setChannelTransacted(transacted);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
assertThat(ReflectionTestUtils.getField(container, "concurrentConsumers")).isEqualTo(2);
container.start();
@@ -115,11 +115,12 @@ public class SimpleMessageListenerContainerLongTests {
}
@Test
public void testAddQueuesAndStartInCycle() throws Exception {
public void testAddQueuesAndStartInCycle() {
final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(
this.connectionFactory);
container.setMessageListener((MessageListener) message -> { });
container.setMessageListener(message -> { });
container.setConcurrentConsumers(2);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
RabbitAdmin admin = new RabbitAdmin(this.connectionFactory);
@@ -145,6 +146,7 @@ public class SimpleMessageListenerContainerLongTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setStartConsumerMinInterval(100);
container.setConsecutiveActiveTrigger(1);
container.setReceiveTimeout(10);
container.setMessageListener(m -> {
try {
Thread.sleep(50);
@@ -184,6 +186,7 @@ public class SimpleMessageListenerContainerLongTests {
container.setQueueNames(QUEUE3);
container.setConcurrentConsumers(2);
container.setMaxConcurrentConsumers(3);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
RabbitTemplate template = new RabbitTemplate(this.connectionFactory);
@@ -212,6 +215,7 @@ public class SimpleMessageListenerContainerLongTests {
container.setQueueNames(QUEUE4);
container.setConcurrentConsumers(2);
container.setMaxConcurrentConsumers(3);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
RabbitTemplate template = new RabbitTemplate(this.connectionFactory);

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
View File

@@ -121,6 +121,7 @@ public class SimpleMessageListenerContainerTests {
container.setQueueNames("foo");
container.setChannelTransacted(false);
container.setTransactionManager(new TestTransactionManager());
container.setReceiveTimeout(10);
container.afterPropertiesSet();
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
container.stop();
@@ -136,6 +137,7 @@ public class SimpleMessageListenerContainerTests {
container.setChannelTransacted(false);
container.setAcknowledgeMode(AcknowledgeMode.NONE);
container.setTransactionManager(new TestTransactionManager());
container.setReceiveTimeout(10);
assertThatIllegalStateException()
.isThrownBy(container::afterPropertiesSet);
container.stop();
@@ -150,6 +152,7 @@ public class SimpleMessageListenerContainerTests {
container.setQueueNames("foo");
container.setChannelTransacted(true);
container.setAcknowledgeMode(AcknowledgeMode.NONE);
container.setReceiveTimeout(10);
assertThatIllegalStateException()
.isThrownBy(container::afterPropertiesSet);
container.stop();
@@ -164,6 +167,7 @@ public class SimpleMessageListenerContainerTests {
container.setQueueNames("foo");
container.setAutoStartup(false);
container.setShutdownTimeout(0);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
assertThat(ReflectionTestUtils.getField(container, "concurrentConsumers")).isEqualTo(1);
container.stop();
@@ -214,6 +218,7 @@ public class SimpleMessageListenerContainerTests {
final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setQueueNames("foo");
container.setBatchSize(2);
container.setReceiveTimeout(10);
container.setMessageListener(messages::add);
container.start();
BasicProperties props = new BasicProperties();
@@ -266,6 +271,7 @@ public class SimpleMessageListenerContainerTests {
final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setQueueNames("foobar");
container.setBatchSize(2);
container.setReceiveTimeout(10);
container.setMessageListener(messages::add);
container.setShutdownTimeout(0);
container.afterPropertiesSet();
@@ -316,6 +322,7 @@ public class SimpleMessageListenerContainerTests {
});
container.setConsumerArguments(Collections.singletonMap("x-priority", 10));
container.setShutdownTimeout(0);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
container.start();
verify(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
@@ -368,6 +375,7 @@ public class SimpleMessageListenerContainerTests {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setQueueNames("foo");
container.setReceiveTimeout(10);
List<?> queues = TestUtils.getPropertyValue(container, "queues", List.class);
assertThat(queues).hasSize(1);
container.addQueueNames(new AnonymousQueue().getName(), new AnonymousQueue().getName());
@@ -399,6 +407,7 @@ public class SimpleMessageListenerContainerTests {
container.setMessageListener(message -> {
});
container.setShutdownTimeout(0);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
for (int i = 0; i < 10; i++) {
@@ -486,6 +495,7 @@ public class SimpleMessageListenerContainerTests {
container.setConcurrentConsumers(2);
container.setQueueNames("foo");
container.setConsumeDelay(100);
container.setReceiveTimeout(10);
container.afterPropertiesSet();
CountDownLatch latch1 = new CountDownLatch(2);
@@ -518,7 +528,7 @@ public class SimpleMessageListenerContainerTests {
waitForConsumersToStop(consumers);
Set<?> allocatedConnections = TestUtils.getPropertyValue(ccf, "allocatedConnections", Set.class);
assertThat(allocatedConnections).hasSize(2);
assertThat(ccf.getCacheProperties().get("openConnections")).isEqualTo("2");
assertThat(ccf.getCacheProperties().get("openConnections")).isEqualTo("1");
}
@Test
@@ -596,6 +606,7 @@ public class SimpleMessageListenerContainerTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames("foo");
container.setReceiveTimeout(10);
container.setPossibleAuthenticationFailureFatal(false);
container.start();
@@ -744,6 +755,7 @@ public class SimpleMessageListenerContainerTests {
container.setMessageListener(listener);
container.setBatchSize(2);
container.setConsumerBatchEnabled(true);
container.setReceiveTimeout(10);
container.start();
BasicProperties props = new BasicProperties();
byte[] payload = "baz".getBytes();

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerWithRabbitMQ.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2024 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.
@@ -63,6 +63,7 @@ public final class SimpleMessageListenerWithRabbitMQ {
container.setBatchSize(500);
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setConcurrentConsumers(20);
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(new SimpleAdapter(), messageConverter));
container.start();

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2024 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.
@@ -106,6 +106,7 @@ public class AmqpAppenderConfiguration implements DisposableBean {
// container.setMessageListener(testListener(4));
container.setAutoStartup(false);
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setReceiveTimeout(10);
return container;
}

7
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2021 the original author or authors.
* Copyright 2002-2024 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.
@@ -96,6 +96,7 @@ public class MissingIdRetryTests {
container.setStatefulRetryFatalWithNullMessageId(false);
container.setMessageListener(new MessageListenerAdapter(new POJO()));
container.setQueueNames("retry.test.queue");
container.setReceiveTimeout(10);
StatefulRetryOperationsInterceptorFactoryBean fb = new StatefulRetryOperationsInterceptorFactoryBean();
@@ -134,6 +135,7 @@ public class MissingIdRetryTests {
ConnectionFactory connectionFactory = ctx.getBean(ConnectionFactory.class);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setPrefetchCount(1);
container.setReceiveTimeout(10);
container.setMessageListener(new MessageListenerAdapter(new POJO()));
container.setQueueNames("retry.test.queue");
@@ -197,6 +199,7 @@ public class MissingIdRetryTests {
}
});
container.setQueueNames("retry.test.queue");
container.setReceiveTimeout(10);
StatefulRetryOperationsInterceptorFactoryBean fb = new StatefulRetryOperationsInterceptorFactoryBean();
@@ -221,7 +224,7 @@ public class MissingIdRetryTests {
try {
assertThat(cdl.await(30, TimeUnit.SECONDS)).isTrue();
Map map = (Map) new DirectFieldAccessor(cache).getPropertyValue("map");
await().until(() -> map.size() == 0);
await().until(map::isEmpty);
ArgumentCaptor putCaptor = ArgumentCaptor.forClass(Object.class);
ArgumentCaptor getCaptor = ArgumentCaptor.forClass(Object.class);
ArgumentCaptor removeCaptor = ArgumentCaptor.forClass(Object.class);

3
spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2023 the original author or authors.
* Copyright 2018-2024 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.
@@ -93,6 +93,7 @@ class EnableRabbitKotlinTests {
fun rabbitListenerContainerFactory(cf: CachingConnectionFactory) =
SimpleRabbitListenerContainerFactory().also {
it.setAcknowledgeMode(AcknowledgeMode.MANUAL)
it.setReceiveTimeout(10)
it.setConnectionFactory(cf)
}