From ca51ca62a6e0ac86c12de8ea805e2b3fba65c960 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 5 Feb 2024 12:49:44 -0500 Subject: [PATCH] SMLC: defer counter release until fetched messages are processed 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`** --- .../examples/TestRabbitTemplateTests.java | 3 +- .../listener/BlockingQueueConsumer.java | 2 +- .../amqp/rabbit/AsyncRabbitTemplateTests.java | 16 +++++----- .../AbstractRabbitAnnotationDrivenTests.java | 20 +++++++------ .../EnableRabbitIntegrationTests.java | 13 ++++---- .../annotation/MockMultiRabbitTests.java | 3 +- ...tenerAnnotationBeanPostProcessorTests.java | 4 ++- .../LocalizedQueueConnectionFactoryTests.java | 11 +++---- .../RoutingConnectionFactoryTests.java | 5 ++-- .../core/BatchingRabbitTemplateTests.java | 22 +++++++------- .../core/FixedReplyQueueDeadLetterTests.java | 5 +++- .../core/RabbitTemplateIntegrationTests.java | 15 ++++++---- .../RabbitTemplateMPPIntegrationTests.java | 3 +- ...atePublisherCallbacksIntegrationTests.java | 30 +++++++++---------- .../BrokerDeclaredQueueNameTests.java | 5 ++-- .../rabbit/listener/ContainerAdminTests.java | 3 +- .../ContainerInitializationTests.java | 3 +- .../listener/ContainerShutDownTests.java | 2 ++ .../JavaConfigFixedReplyQueueTests.java | 5 +++- .../listener/LocallyTransactedSMLCTests.java | 5 ++-- ...ContainerErrorHandlerIntegrationTests.java | 5 ++-- ...nerContainerLifecycleIntegrationTests.java | 23 ++++++++------ ...ontainerMultipleQueueIntegrationTests.java | 7 +++-- ...istenerContainerRetryIntegrationTests.java | 6 ++-- .../MessageListenerContainerTxSynchTests.java | 3 +- ...sageListenerManualAckIntegrationTests.java | 1 + ...MessageListenerTxSizeIntegrationTests.java | 6 ++-- .../listener/QueueDeclarationTests.java | 3 +- ...ageListenerContainerIntegration2Tests.java | 8 ++++- ...mpleMessageListenerContainerLongTests.java | 12 +++++--- .../SimpleMessageListenerContainerTests.java | 14 ++++++++- .../SimpleMessageListenerWithRabbitMQ.java | 3 +- .../logback/AmqpAppenderConfiguration.java | 3 +- .../rabbit/retry/MissingIdRetryTests.java | 7 +++-- .../annotation/EnableRabbitKotlinTests.kt | 3 +- 35 files changed, 174 insertions(+), 105 deletions(-) diff --git a/spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/examples/TestRabbitTemplateTests.java b/spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/examples/TestRabbitTemplateTests.java index 80631323..4d8c6958 100644 --- a/spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/examples/TestRabbitTemplateTests.java +++ b/spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/examples/TestRabbitTemplateTests.java @@ -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") diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index b78c21c2..4e6048b2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java index 039355ec..4b03c741 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java @@ -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 = diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java index 14bbb118..271731d2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 59808981..210af62d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -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 amqp656 = await().until(() -> queueInfo("amqp656"), q -> q != null); + Map 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 errorHandlerError() { - return new AtomicReference(); + return new AtomicReference<>(); } @Bean diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MockMultiRabbitTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MockMultiRabbitTests.java index 3aa30d58..cda0362f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MockMultiRabbitTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MockMultiRabbitTests.java @@ -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(); }); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java index 29b65189..6ef00c59 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java index 9dc9246b..29192e6a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java @@ -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 channels = new HashMap(); + private final Map channels = new HashMap<>(); - private final Map consumers = new HashMap(); + private final Map consumers = new HashMap<>(); - private final Map consumerTags = new HashMap(); + private final Map 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 mockCFs = new HashMap(); + final Map 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 captor = ArgumentCaptor.forClass(String.class); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(lqcf); + container.setReceiveTimeout(10); container.setQueueNames("q"); container.afterPropertiesSet(); container.start(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java index c22f4b6e..42bec349 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java @@ -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 factories = new HashMap(2); + Map 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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java index e623ed04..e5587a97 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java @@ -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 received = new ArrayList(); + final List 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 received = new ArrayList(); + final List 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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java index f4dbce98..1b357e72 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java @@ -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; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 647e931b..00dea012 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -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 results = new ConcurrentHashMap(); + final Map 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 replyToWas = new AtomicReference(); + container.setReceiveTimeout(10); + final AtomicReference 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()); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java index dd5a85b0..97f37a3e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 2c9808fe..9c158ebd 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -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 confirmCD = new AtomicReference(); + final AtomicReference 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::toUpperCase)); container.start(); @@ -285,7 +287,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { @Test public void testPublisherReturns() throws Exception { final CountDownLatch latch = new CountDownLatch(1); - final List returns = new ArrayList(); + final List 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 returns = new ArrayList(); + final List 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 confirms = new HashSet(); + final Set 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 correlation = new AtomicReference(); - final AtomicReference reason = new AtomicReference(); + final AtomicReference correlation = new AtomicReference<>(); + final AtomicReference 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 log = new AtomicReference(); + final AtomicReference 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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BrokerDeclaredQueueNameTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BrokerDeclaredQueueNameTests.java index d65698a3..434621ee 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BrokerDeclaredQueueNameTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BrokerDeclaredQueueNameTests.java @@ -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; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerAdminTests.java index 62f47887..22ef6ae4 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerAdminTests.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java index 88d03d1f..c0b87bed 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java @@ -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") diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java index db23cde2..1902921a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java @@ -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); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java index 7f5056fe..d795411f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/JavaConfigFixedReplyQueueTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedSMLCTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedSMLCTests.java index 043679f7..60ebe309 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedSMLCTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedSMLCTests.java @@ -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; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java index 627c8a76..cba98cee 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java @@ -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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index 6aeb1ea4..cb53e80b 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java @@ -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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java index 343fc7ff..155eeaa8 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java index 56635753..576ac75a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java @@ -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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java index fb097e44..cc0b9f8c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java @@ -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"); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java index beb265e8..5d855006 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java index adc1332c..7a9df1b4 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java @@ -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 count = new ThreadLocal(); + private final ThreadLocal count = new ThreadLocal<>(); private final CountDownLatch latch; @@ -174,6 +175,7 @@ public class MessageListenerTxSizeIntegrationTests { latch.countDown(); } } + } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/QueueDeclarationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/QueueDeclarationTests.java index 2c99f8a8..b963918c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/QueueDeclarationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/QueueDeclarationTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 0c247fc9..59ae41eb 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java @@ -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 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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java index 81faf4f8..0a0a1641 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java @@ -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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java index 7044f5ee..5324bbcf 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerWithRabbitMQ.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerWithRabbitMQ.java index ad3d808d..1de168c5 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerWithRabbitMQ.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerWithRabbitMQ.java @@ -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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderConfiguration.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderConfiguration.java index 2e460570..17275801 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderConfiguration.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderConfiguration.java @@ -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; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java index eef04a3a..860a4af4 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java @@ -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); diff --git a/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt b/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt index 8eadd203..ea320a05 100644 --- a/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt +++ b/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt @@ -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) }