GH-290: Add Consumer-side Batching

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/290
Resolves #304

docs/src/main/asciidoc/overview.adoc

@@ -424,9 +424,188 @@ This allows you to add arguments that are not currently directly supported by th
 [[rabbit-receiving-batch]]
 === Receiving Batched Messages
 
+With the RabbitMQ binder, there are two types of batches handled by consumer bindings:
+
+==== Batches Created by Producers
+
 Normally, if a producer binding has `batch-enabled=true` (see <<rabbit-prod-props>>), or a message is created by a `BatchingRabbitTemplate`, elements of the batch are returned as individual calls to the listener method.
 Starting with version 3.0, any such batch can be presented as a `List<?>` to the listener method if `spring.cloud.stream.bindings.<name>.consumer.batch-mode` is set to `true`.
 
+==== Consumer-side Batching
+
+Starting with version 3.1, the consumer can be configured to assemble multiple inbound messages into a batch which is presented to the application as a `List<?>` of converted payloads.
+The following simple application demonstrates how to use this technique:
+
+====
+[source, properties]
+----
+spring.cloud.stream.bindings.input-in-0.group=someGroup
+
+spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true
+
+spring.cloud.stream.rabbit.bindings.input-in-0.consumer.enable-batching=true
+spring.cloud.stream.rabbit.bindings.input-in-0.consumer.batch-size=10
+spring.cloud.stream.rabbit.bindings.input-in-0.consumer.receive-timeout=200
+----
+====
+
+====
+[source, java]
+----
+@SpringBootApplication
+public class Application {
+
+	public static void main(String[] args) {
+		SpringApplication.run(Application.class, args);
+	}
+
+	@Bean
+	Consumer<List<Thing>> input() {
+		return list -> {
+			System.out.println("Received " + list.size());
+			list.forEach(thing -> {
+				System.out.println(thing);
+
+				// ...
+
+			});
+		};
+	}
+
+	@Bean
+	public ApplicationRunner runner(RabbitTemplate template) {
+		return args -> {
+			template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value1\"}");
+			template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value2\"}");
+		};
+	}
+
+	public static class Thing {
+
+		private String field;
+
+		public Thing() {
+		}
+
+		public Thing(String field) {
+			this.field = field;
+		}
+
+		public String getField() {
+			return this.field;
+		}
+
+		public void setField(String field) {
+			this.field = field;
+		}
+
+		@Override
+		public String toString() {
+			return "Thing [field=" + this.field + "]";
+		}
+
+	}
+
+}
+----
+====
+
+====
+[source]
+----
+Received 2
+Thing [field=value1]
+Thing [field=value2]
+----
+====
+
+The number of messages in a batch is specified by the `batch-size` and `receive-timeout` properties; if the `receive-timeout` elapses with no new messages, a "short" batch is delivered.
+
+IMPORTANT: Consumer-side batching is only supported with `container-type=simple` (the default).
+
+If you wish to examine headers of consumer-side batched messages, you should consume `Message<List<?>>`; the headers are a `List<Map<String, Object>>` in a header `AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS`, with the headers for each payload element in the corresponding index.
+Again, here is a simple example:
+
+====
+[source, java]
+----
+@SpringBootApplication
+public class Application {
+
+	public static void main(String[] args) {
+		SpringApplication.run(Application.class, args);
+	}
+
+	@Bean
+	Consumer<Message<List<Thing>>> input() {
+		return msg -> {
+			List<Thing> things = msg.getPayload();
+			System.out.println("Received " + things.size());
+			@SuppressWarnings("unchecked")
+			List<Map<String, Object>> headers =
+					(List<Map<String, Object>>) msg.getHeaders().get(AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS);
+			for (int i = 0; i < things.size(); i++) {
+				System.out.println(things.get(i) + " myHeader=" + headers.get(i).get("myHeader"));
+
+				// ...
+
+			}
+		};
+	}
+
+	@Bean
+	public ApplicationRunner runner(RabbitTemplate template) {
+		return args -> {
+			template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value1\"}", msg -> {
+				msg.getMessageProperties().setHeader("myHeader", "headerValue1");
+				return msg;
+			});
+			template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value2\"}", msg -> {
+				msg.getMessageProperties().setHeader("myHeader", "headerValue2");
+				return msg;
+			});
+		};
+	}
+
+	public static class Thing {
+
+		private String field;
+
+		public Thing() {
+		}
+
+		public Thing(String field) {
+			this.field = field;
+		}
+
+		public String getfield() {
+			return this.field;
+		}
+
+		public void setfield(String field) {
+			this.field = field;
+		}
+
+		@Override
+		public String toString() {
+			return "Thing [field=" + this.field + "]";
+		}
+
+	}
+
+}
+----
+====
+
+====
+[source]
+----
+Received 2
+Thing [field=value1] myHeader=headerValue1
+Thing [field=value2] myHeader=headerValue2
+----
+====
+
 [[rabbit-prod-props]]
 === Rabbit Producer Properties
 

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java

@@ -125,6 +125,21 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	 */
 	private String anonymousGroupPrefix = "anonymous.";
 
+	/**
+	 * When true, the listener container will assemble a list from multiple messages,
+	 * according to the batchSize and receiveTimeout properties. Only applies with
+	 * {@link ContainerType#SIMPLE}.
+	 */
+	private boolean enableBatching;
+
+	/**
+	 * How long to block waiting to receive messages; increasing from the default 1 second
+	 * will make the binding less responsive to stop requests. When enableConsumerBatching
+	 * is true, a short batch may be emitted if this time elapses before the batchSize is
+	 * satisfied. Only applies with {@link ContainerType#SIMPLE}.
+	 */
+	private Long receiveTimeout;
+
 	public boolean isTransacted() {
 		return transacted;
 	}
@@ -317,4 +332,20 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 		this.anonymousGroupPrefix = anonymousGroupPrefix;
 	}
 
+	public boolean isEnableBatching() {
+		return this.enableBatching;
+	}
+
+	public void setEnableBatching(boolean enableBatching) {
+		this.enableBatching = enableBatching;
+	}
+
+	public Long getReceiveTimeout() {
+		return this.receiveTimeout;
+	}
+
+	public void setReceiveTimeout(Long receiveTimeout) {
+		this.receiveTimeout = receiveTimeout;
+	}
+
 }

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -88,6 +88,7 @@ import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.integration.acks.AcknowledgmentCallback.Status;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.BatchMode;
 import org.springframework.integration.amqp.inbound.AmqpMessageSource;
 import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
 import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
@@ -446,16 +447,17 @@ public class RabbitMessageChannelBinder extends
 		Assert.state(!HeaderMode.embeddedHeaders.equals(properties.getHeaderMode()),
 				"the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively");
 		String destination = consumerDestination.getName();
-		boolean directContainer = properties.getExtension().getContainerType()
+		RabbitConsumerProperties extension = properties.getExtension();
+		boolean directContainer = extension.getContainerType()
 				.equals(ContainerType.DIRECT);
 		AbstractMessageListenerContainer listenerContainer = directContainer
 				? new DirectMessageListenerContainer(this.connectionFactory)
 				: new SimpleMessageListenerContainer(this.connectionFactory);
 		listenerContainer
-				.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
-		listenerContainer.setChannelTransacted(properties.getExtension().isTransacted());
+				.setAcknowledgeMode(extension.getAcknowledgeMode());
+		listenerContainer.setChannelTransacted(extension.isTransacted());
 		listenerContainer
-				.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
+				.setDefaultRequeueRejected(extension.isRequeueRejected());
 		int concurrency = properties.getConcurrency();
 		concurrency = concurrency > 0 ? concurrency : 1;
 		if (directContainer) {
@@ -466,9 +468,9 @@ public class RabbitMessageChannelBinder extends
 			setSMLCProperties(properties,
 					(SimpleMessageListenerContainer) listenerContainer, concurrency);
 		}
-		listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
+		listenerContainer.setPrefetchCount(extension.getPrefetch());
 		listenerContainer
-				.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
+				.setRecoveryInterval(extension.getRecoveryInterval());
 		listenerContainer.setTaskExecutor(
 				new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
 		String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true);
@@ -476,12 +478,12 @@ public class RabbitMessageChannelBinder extends
 		listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
 		listenerContainer.setMessagePropertiesConverter(
 				RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
-		listenerContainer.setExclusive(properties.getExtension().isExclusive());
+		listenerContainer.setExclusive(extension.isExclusive());
 		listenerContainer
-				.setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal());
-		if (properties.getExtension().getFailedDeclarationRetryInterval() != null) {
+				.setMissingQueuesFatal(extension.getMissingQueuesFatal());
+		if (extension.getFailedDeclarationRetryInterval() != null) {
 			listenerContainer.setFailedDeclarationRetryInterval(
-					properties.getExtension().getFailedDeclarationRetryInterval());
+					extension.getFailedDeclarationRetryInterval());
 		}
 		if (getApplicationEventPublisher() != null) {
 			listenerContainer
@@ -492,10 +494,10 @@ public class RabbitMessageChannelBinder extends
 		}
 		getContainerCustomizer().configure(listenerContainer,
 				consumerDestination.getName(), group);
-		if (StringUtils.hasText(properties.getExtension().getConsumerTagPrefix())) {
+		if (StringUtils.hasText(extension.getConsumerTagPrefix())) {
 			final AtomicInteger index = new AtomicInteger();
 			listenerContainer.setConsumerTagStrategy(
-					q -> properties.getExtension().getConsumerTagPrefix() + "#"
+					q -> extension.getConsumerTagPrefix() + "#"
 							+ index.getAndIncrement());
 		}
 		listenerContainer.afterPropertiesSet();
@@ -506,7 +508,7 @@ public class RabbitMessageChannelBinder extends
 		adapter.setBeanFactory(this.getBeanFactory());
 		adapter.setBeanName("inbound." + destination);
 		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
-		mapper.setRequestHeaderNames(properties.getExtension().getHeaderPatterns());
+		mapper.setRequestHeaderNames(extension.getHeaderPatterns());
 		adapter.setHeaderMapper(mapper);
 		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(
 				consumerDestination, group, properties);
@@ -519,6 +521,10 @@ public class RabbitMessageChannelBinder extends
 			adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
 		}
 		adapter.setMessageConverter(passThoughConverter);
+		if (properties.isBatchMode() && extension.isEnableBatching()
+				&& ContainerType.SIMPLE.equals(extension.getContainerType())) {
+			adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS);
+		}
 		return adapter;
 	}
 
@@ -526,16 +532,24 @@ public class RabbitMessageChannelBinder extends
 			ExtendedConsumerProperties<RabbitConsumerProperties> properties,
 			SimpleMessageListenerContainer listenerContainer, int concurrency) {
 
+		RabbitConsumerProperties extension = properties.getExtension();
 		listenerContainer.setConcurrentConsumers(concurrency);
-		int maxConcurrency = properties.getExtension().getMaxConcurrency();
+		int maxConcurrency = extension.getMaxConcurrency();
 		if (maxConcurrency > concurrency) {
 			listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
 		}
 		listenerContainer.setDeBatchingEnabled(!properties.isBatchMode());
-		listenerContainer.setBatchSize(properties.getExtension().getBatchSize());
-		if (properties.getExtension().getQueueDeclarationRetries() != null) {
+		listenerContainer.setBatchSize(extension.getBatchSize());
+		if (extension.getQueueDeclarationRetries() != null) {
 			listenerContainer.setDeclarationRetries(
-					properties.getExtension().getQueueDeclarationRetries());
+					extension.getQueueDeclarationRetries());
+		}
+		if (properties.isBatchMode() && extension.isEnableBatching()) {
+			listenerContainer.setConsumerBatchEnabled(true);
+			listenerContainer.setDeBatchingEnabled(true);
+		}
+		if (extension.getReceiveTimeout() != null) {
+			listenerContainer.setReceiveTimeout(extension.getReceiveTimeout());
 		}
 	}
 

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -1557,7 +1557,7 @@ public class RabbitBinderTests extends
 
 	@SuppressWarnings("unchecked")
 	@Test
-	public void testConsumerBatching() throws Exception {
+	public void testProducerBatching() throws Exception {
 		RabbitTestBinder binder = getBinder();
 		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.getExtension()
@@ -1570,7 +1570,44 @@ public class RabbitBinderTests extends
 
 		DirectChannel output = createBindableChannel("output",
 				createProducerBindingProperties(producerProperties));
-		output.setBeanName("consumerBatchingProducer");
+		output.setBeanName("producerBatchingProducer");
+		Binding<MessageChannel> producerBinding = binder.bindProducer("p.batching.0",
+				output, producerProperties);
+
+		QueueChannel input = new QueueChannel();
+		input.setBeanName("producerBatchingConsumer");
+		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.setBatchMode(true);
+		consumerProperties.getExtension().setBatchSize(2);
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("p.batching.0",
+				"producerBatching", input, consumerProperties);
+		output.send(new GenericMessage<>("foo".getBytes()));
+		output.send(new GenericMessage<>("bar".getBytes()));
+
+
+		Message<?> in = input.receive(10000);
+		assertThat(in).isNotNull();
+		assertThat(in.getPayload()).isInstanceOf(List.class);
+		List<byte[]> payload = (List<byte[]>) in.getPayload();
+		assertThat(payload).hasSize(2);
+		assertThat(payload.get(0)).isEqualTo("foo".getBytes());
+		assertThat(payload.get(1)).isEqualTo("bar".getBytes());
+
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testConsumerBatching() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
+		producerProperties.getExtension()
+				.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+
+		DirectChannel output = createBindableChannel("output",
+				createProducerBindingProperties(producerProperties));
+		output.setBeanName("consumerBatching.Producer");
 		Binding<MessageChannel> producerBinding = binder.bindProducer("c.batching.0",
 				output, producerProperties);
 
@@ -1579,6 +1616,7 @@ public class RabbitBinderTests extends
 		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setBatchMode(true);
 		consumerProperties.getExtension().setBatchSize(2);
+		consumerProperties.getExtension().setEnableBatching(true);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("c.batching.0",
 				"consumerBatching", input, consumerProperties);
 		output.send(new GenericMessage<>("foo".getBytes()));