GH-1218: Backport Schema Fixes

Resolves #1218

- batch properties

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -54,6 +54,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
 
 	private static final String PUBLISHER_CONFIRMS = "publisher-confirms";
 
+	private static final String CONFIRM_TYPE = "confirm-type";
+
 	private static final String PUBLISHER_RETURNS = "publisher-returns";
 
 	private static final String REQUESTED_HEARTBEAT = "requested-heartbeat";
@@ -114,7 +116,7 @@ protected void doParse(Element element, ParserContext parserContext, BeanDefinit
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, FACTORY_TIMEOUT, "channelCheckoutTimeout");
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, CONNECTION_LIMIT);
 		NamespaceUtils.setReferenceIfAttributeDefined(builder, element, "connection-name-strategy");
-		NamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-type", "publisherConfirmType");
+		NamespaceUtils.setValueIfAttributeDefined(builder, element, CONFIRM_TYPE, "publisherConfirmType");
 	}
 
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -74,6 +74,10 @@ public final class RabbitNamespaceUtils {
 
 	private static final String TRANSACTION_SIZE_ATTRIBUTE = "transaction-size";
 
+	private static final String CONSUMER_BATCH_ENABLED_ATTRIBUTE = "consumer-batch-enabled";
+
+	private static final String BATCH_SIZE_ATTRIBUTE = "batch-size";
+
 	private static final String PHASE_ATTRIBUTE = "phase";
 
 	private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup";
@@ -216,9 +220,23 @@ public static BeanDefinition parseContainer(Element containerEle, ParserContext
 			containerDef.getPropertyValues().add("channelTransacted", new TypedStringValue(channelTransacted));
 		}
 
+		String consumerBatch = containerEle.getAttribute(CONSUMER_BATCH_ENABLED_ATTRIBUTE);
+		if (StringUtils.hasText(consumerBatch)) {
+			containerDef.getPropertyValues().add("consumerBatchEnabled", new TypedStringValue(consumerBatch));
+		}
+
+		String batchSize = containerEle.getAttribute(BATCH_SIZE_ATTRIBUTE);
+		if (StringUtils.hasText(batchSize)) {
+			containerDef.getPropertyValues().add("batchSize", new TypedStringValue(batchSize));
+		}
+
 		String transactionSize = containerEle.getAttribute(TRANSACTION_SIZE_ATTRIBUTE);
 		if (StringUtils.hasText(transactionSize)) {
-			containerDef.getPropertyValues().add("txSize", new TypedStringValue(transactionSize));
+			if (StringUtils.hasText(batchSize)) {
+				parserContext.getReaderContext().error(
+						"Listener Container - cannot have both 'batch-size' and 'transaction-size'", containerEle);
+			}
+			containerDef.getPropertyValues().add("batchSize", new TypedStringValue(transactionSize));
 		}
 
 		String requeueRejected = containerEle.getAttribute(REQUEUE_REJECTED_ATTRIBUTE);

spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd

@@ -736,20 +736,40 @@
 		<xsd:attribute name="transaction-size" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
+	Deprecated. Synonym to 'batch-size'.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="consumer-batch-enabled" default="false">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	Enable consumer-side batching according to 'batch-size' and 'receive-timeout'.
+	Only applies when the container type is 'simple'.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string" />
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="batch-size" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
 	Tells the container how many messages to process in a single transaction (if the channel is transactional). For
-	best results it should be less than or equal to the prefetch count. Also used to determine how often acks
-	are sent when using AUTO acknowledge mode.
+	best results it should be less than or equal to the prefetch count.
+	Also used to determine how often acks are sent when using AUTO acknowledge mode.
+	Also used to determine the batch size when 'consumer-batch-enabled' is 'true'.
 	Only applies when the container type is 'simple'.
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attribute name="receive-timeout" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
-	The time for which a consumer waits for a message; used together with 'transaction-size' to determine whether
-	or not a consumer is idle, why dynamic concurrency is enabled using 'max-concurrency'. When 'transaction-size'
-	is greater than 1 (default), acks for processed message(s) can be delayed up to ('transaction-size' - 1) *
-	this value because the ack is sent after `transaction-size` attempts to receive a message have occurred.
+	The time for which a consumer waits for a message; used together with 'batch-size' to determine whether
+	or not a consumer is idle, why dynamic concurrency is enabled using 'max-concurrency'. When 'batch-size'
+	is greater than 1 (default), acks for processed message(s) can be delayed up to ('batch-size' - 1) *
+	this value because the ack is sent after 'batch-size' attempts to receive a message have occurred.
+	Also used to deliver a short batch when 'consumer-batch-enabled' is 'true'.
 	Only applies when the container type is 'simple'.
 				]]></xsd:documentation>
 			</xsd:annotation>

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2019 the original author or authors.
+ * Copyright 2010-2020 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.
@@ -180,6 +180,7 @@ public void testParseWithTx() {
 		SimpleMessageListenerContainer container = beanFactory.getBean("container6", SimpleMessageListenerContainer.class);
 		assertThat(container.isChannelTransacted()).isTrue();
 		assertThat(ReflectionTestUtils.getField(container, "batchSize")).isEqualTo(5);
+		assertThat(ReflectionTestUtils.getField(container, "consumerBatchEnabled")).isEqualTo(Boolean.TRUE);
 	}
 
 	@Test

spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml

@@ -40,7 +40,8 @@
 		<rabbit:listener id="container5" queues="foo" ref="testBean" method="handle"/>
 	</rabbit:listener-container>
 
-	<rabbit:listener-container connection-factory="connectionFactory" channel-transacted="true" transaction-size="5" >
+	<rabbit:listener-container connection-factory="connectionFactory" channel-transacted="true" batch-size="5"
+			consumer-batch-enabled="true" >
 		<rabbit:listener id="container6" queues="foo" ref="testBean" method="handle"/>
 	</rabbit:listener-container>