diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java index a82840e6..2e8c5afd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2022 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. @@ -123,13 +123,18 @@ public class SimpleRabbitListenerContainerFactory /** * Set to true to present a list of messages based on the {@link #setBatchSize(Integer)}, - * if the listener supports it. + * if the listener supports it. Starting with version 3.0, setting this to true will + * also {@link #setBatchListener(boolean)} to true. * @param consumerBatchEnabled true to create message batches in the container. * @since 2.2 * @see #setBatchSize(Integer) + * @see #setBatchListener(boolean) */ public void setConsumerBatchEnabled(boolean consumerBatchEnabled) { this.consumerBatchEnabled = consumerBatchEnabled; + if (consumerBatchEnabled) { + setBatchListener(true); + } } @Override diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index a409ec42..41c3ebdc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -20,6 +20,7 @@ import java.lang.reflect.Method; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.lang.reflect.WildcardType; +import java.util.Collection; import java.util.List; import org.springframework.amqp.core.MessageProperties; @@ -328,6 +329,8 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis private boolean isAmqpMessageList; + private boolean isCollection; + MessagingMessageConverterAdapter(Object bean, Method method, boolean batch) { this.bean = bean; this.method = method; @@ -392,6 +395,12 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis if (genericParameterType == null) { genericParameterType = extractGenericParameterTypFromMethodParameter(methodParameter); + if (this.isBatch && !this.isCollection) { + throw new IllegalStateException( + "Mis-configuration; a batch listener must consume a List or " + + "Collection for method: " + this.method); + } + } else { if (MessagingMessageListenerAdapter.this.logger.isDebugEnabled()) { @@ -435,9 +444,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis genericParameterType = ((ParameterizedType) genericParameterType).getActualTypeArguments()[0]; } else if (this.isBatch - && parameterizedType.getRawType().equals(List.class) - && parameterizedType.getActualTypeArguments().length == 1) { + && ((parameterizedType.getRawType().equals(List.class) + || parameterizedType.getRawType().equals(Collection.class)) + && parameterizedType.getActualTypeArguments().length == 1)) { + this.isCollection = true; Type paramType = parameterizedType.getActualTypeArguments()[0]; boolean messageHasGeneric = paramType instanceof ParameterizedType && ((ParameterizedType) paramType).getRawType().equals(Message.class); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java index 6837be51..4bcb6f92 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java @@ -19,6 +19,8 @@ package org.springframework.amqp.rabbit.annotation; import static org.assertj.core.api.Assertions.assertThat; import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -51,7 +53,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; */ @SpringJUnitConfig @DirtiesContext -@RabbitAvailable(queues = { "batch.1", "batch.2", "batch.3", "batch.4" }) +@RabbitAvailable(queues = { "batch.1", "batch.2", "batch.3", "batch.4", "batch.5" }) public class EnableRabbitBatchIntegrationTests { @Autowired @@ -114,6 +116,16 @@ public class EnableRabbitBatchIntegrationTests { .isEqualTo(2); } + @Test + public void collectionWithStringInfer() throws InterruptedException { + this.template.convertAndSend("batch.5", new Foo("foo")); + this.template.convertAndSend("batch.5", new Foo("bar")); + assertThat(this.listener.fivesLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.fives).hasSize(2); + assertThat(this.listener.fives.get(0).getBar()).isEqualTo("foo"); + assertThat(this.listener.fives.get(1).getBar()).isEqualTo("bar"); + } + @Configuration @EnableRabbit public static class Config { @@ -186,6 +198,10 @@ public class EnableRabbitBatchIntegrationTests { CountDownLatch fooConsumerBatchTooLatch = new CountDownLatch(1); + List fives = new ArrayList<>(); + + CountDownLatch fivesLatch = new CountDownLatch(1); + private List nativeMessages; private final CountDownLatch nativeMessagesLatch = new CountDownLatch(1); @@ -214,6 +230,12 @@ public class EnableRabbitBatchIntegrationTests { this.nativeMessagesLatch.countDown(); } + @RabbitListener(queues = "batch.5") + public void listen5(Collection in) { + this.fives.addAll(in); + this.fivesLatch.countDown(); + } + } @SuppressWarnings("serial") diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapterTests.java new file mode 100644 index 00000000..358c4676 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapterTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.listener.adapter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; + +import java.lang.reflect.Method; +import java.util.List; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.utils.test.TestUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class BatchMessagingMessageListenerAdapterTests { + + @Test + void compatibleMethod() throws Exception { + Method method = getClass().getDeclaredMethod("listen", List.class); + BatchMessagingMessageListenerAdapter adapter = new BatchMessagingMessageListenerAdapter(this, method, false, + null, null); + assertThat(TestUtils.getPropertyValue(adapter, "messagingMessageConverter.inferredArgumentType")) + .isEqualTo(String.class); + Method badMethod = getClass().getDeclaredMethod("listen", String.class); + assertThatIllegalStateException().isThrownBy(() -> + new BatchMessagingMessageListenerAdapter(this, badMethod, false, null, null) + ).withMessageStartingWith("Mis-configuration"); + } + + public void listen(String in) { + } + + public void listen(List in) { + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index fae9a422..8cf2ae89 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -3422,7 +3422,7 @@ Adding a `group` attribute causes a bean of type `Collection> of messages, the de-batching is normally performed by the container and the listener is invoked with one message at at time. -Starting with version 2.2, you can configure the listener container factory and listener to receive the entire batch in one call, simply set the factory's `batchListener` property, and make the method payload parameter a `List`: +Starting with version 2.2, you can configure the listener container factory and listener to receive the entire batch in one call, simply set the factory's `batchListener` property, and make the method payload parameter a `List` or `Collection`: ==== [source, java] @@ -3486,20 +3486,17 @@ When using `consumerBatchEnabled` with `@RabbitListener`: ---- @RabbitListener(queues = "batch.1", containerFactory = "consumerBatchContainerFactory") public void consumerBatch1(List amqpMessages) { - this.amqpMessagesReceived = amqpMessages; - this.batch1Latch.countDown(); + ... } @RabbitListener(queues = "batch.2", containerFactory = "consumerBatchContainerFactory") public void consumerBatch2(List> messages) { - this.messagingMessagesReceived = messages; - this.batch2Latch.countDown(); + ... } @RabbitListener(queues = "batch.3", containerFactory = "consumerBatchContainerFactory") public void consumerBatch3(List strings) { - this.batch3Strings = strings; - this.batch3Latch.countDown(); + ... } ---- ==== @@ -3511,6 +3508,12 @@ public void consumerBatch3(List strings) { You can also add a `Channel` parameter, often used when using `MANUAL` ack mode. This is not very useful with the third example because you don't have access to the `delivery_tag` property. +Spring Boot provides a configuration property for `consumerBatchEnabled` and `batchSize`, but not for `batchListener`. +Starting with version 3.0, setting `consumerBatchEnabled` to `true` on the container factory also sets `batchListener` to `true`. +When `consumerBatchEnabled` is `true`, the listener **must** be a batch listener. + +Starting with version 3.0, listener methods can consume `Collection` or `List`. + [[using-container-factories]] ===== Using Container Factories diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 07cfb6f1..b3279436 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -21,3 +21,10 @@ See <> for more information. `RabbitStreamOperations2` and `RabbitStreamTemplate2` have been deprecated in favor of `RabbitStreamOperations` and `RabbitStreamTemplate` respectively. See <> for more information. + +==== `@RabbitListener` Changes + +Batch listeners can now consume `Collection` as well as `List`. +The batch messaging adapter now ensures that the method is suitable for consuming batches. +When setting the container factory `consumerBatchEnabled` to `true`, the `batchListener` property is also set to `true`. +See <> for more infoprmation.