GH-1489: Batch RabbitListener Improvements

Resolves #1489

- allow listeners to consume `Collection<?>` as well as `List<?>`
- detect a non-batch listener method in the batch adapter
- coerce `batchListener` to `true` when `consumerBatchEnabled` is true

Author: garyrussell

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 void setReceiveTimeout(Long receiveTimeout) {
 
 	/**
 	 * 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

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -20,6 +20,7 @@
 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 @@ protected final class MessagingMessageConverterAdapter extends MessagingMessageC
 
 		private boolean isAmqpMessageList;
 
+		private boolean isCollection;
+
 		MessagingMessageConverterAdapter(Object bean, Method method, boolean batch) {
 			this.bean = bean;
 			this.method = method;
@@ -392,6 +395,12 @@ private Type determineInferredType() { // NOSONAR - complexity
 
 					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 @@ private Type extractGenericParameterTypFromMethodParameter(MethodParameter metho
 					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);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java

@@ -19,6 +19,8 @@
 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 @@
  */
 @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 void nativeMessageList() throws InterruptedException {
 				.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 static class Listener {
 
 		CountDownLatch fooConsumerBatchTooLatch = new CountDownLatch(1);
 
+		List<Foo> fives = new ArrayList<>();
+
+		CountDownLatch fivesLatch = new CountDownLatch(1);
+
 		private List<org.springframework.amqp.core.Message> nativeMessages;
 
 		private final CountDownLatch nativeMessagesLatch = new CountDownLatch(1);
@@ -214,6 +230,12 @@ public void listen4(List<org.springframework.amqp.core.Message> in) {
 			this.nativeMessagesLatch.countDown();
 		}
 
+		@RabbitListener(queues = "batch.5")
+		public void listen5(Collection<Foo> in) {
+			this.fives.addAll(in);
+			this.fivesLatch.countDown();
+		}
+
 	}
 
 	@SuppressWarnings("serial")

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<String> in) {
+	}
+
+}

src/reference/asciidoc/amqp.adoc

@@ -3422,7 +3422,7 @@ Adding a `group` attribute causes a bean of type `Collection<MessageListenerCont
 ===== @RabbitListener with Batching
 
 When receiving a <<template-batching, a batch>> 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<Message> amqpMessages) {
-    this.amqpMessagesReceived = amqpMessages;
-    this.batch1Latch.countDown();
+    ...
 }
 
 @RabbitListener(queues = "batch.2", containerFactory = "consumerBatchContainerFactory")
 public void consumerBatch2(List<org.springframework.messaging.Message<Invoice>> messages) {
-    this.messagingMessagesReceived = messages;
-    this.batch2Latch.countDown();
+    ...
 }
 
 @RabbitListener(queues = "batch.3", containerFactory = "consumerBatchContainerFactory")
 public void consumerBatch3(List<Invoice> strings) {
-    this.batch3Strings = strings;
-    this.batch3Latch.countDown();
+    ...
 }
 ----
 ====
@@ -3511,6 +3508,12 @@ public void consumerBatch3(List<Invoice> 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
 

src/reference/asciidoc/whats-new.adoc

@@ -21,3 +21,10 @@ See <<async-template>> for more information.
 
 `RabbitStreamOperations2` and `RabbitStreamTemplate2` have been deprecated in favor of `RabbitStreamOperations` and `RabbitStreamTemplate` respectively.
 See <<stream-support>> 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 <<receiving-batch>> for more infoprmation.