AMQP-451 Batching RabbitTemplate

JIRA: https://jira.spring.io/browse/AMQP-451

Released in stealth mode for now (no docs, javadocs marked `Experimental`).

This feature is to support XD; we can't incubate in XD because it needs a
change to the listener container.

AMQP-451 Add Perf Test

Move Debatching Code

Move from `SMLC.doReceiveAndExecute()` to `AMLC.executeListener()`.

- Change the cause of the `ListenerExecutionFailedException` on a bad
decode to a `MessageConversionException` so that the default error handler
will cause the message to be rejected - otherwise the badly formed message
will continually be redelivered by default.
- Add a test case to ensure a badly formed message is rejected.

spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java

@@ -42,6 +42,11 @@ public class MessageProperties implements Serializable {
 
 	public static final String CONTENT_TYPE_XML = "application/xml";
 
+	public static final String SPRING_BATCH_FORMAT = "springBatchFormat";
+
+	public static final String BATCH_FORMAT_LENGTH_HEADER4 = "lengthHeader4";
+
+
 	static final String DEFAULT_CONTENT_TYPE = CONTENT_TYPE_BYTES;
 
 	static final MessageDeliveryMode DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.core;
+
+import java.util.Date;
+import java.util.concurrent.ScheduledFuture;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.MessageBatch;
+import org.springframework.scheduling.TaskScheduler;
+
+/**
+ * A {@link RabbitTemplate} that permits batching individual messages into a larger
+ * message. All {@code send()} methods (except
+ * {@link #send(String, String, org.springframework.amqp.core.Message,
+ * org.springframework.amqp.rabbit.support.CorrelationData)})
+ * are eligible for batching.
+ * <p>
+ * <b>Experimental - APIs may change.</b>
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class BatchingRabbitTemplate extends RabbitTemplate {
+
+	private final BatchingStrategy batchingStrategy;
+
+	private final TaskScheduler scheduler;
+
+	private volatile ScheduledFuture<?> scheduledTask;
+
+	/**
+	 * @param batchingStrategy the batching strategy.
+	 * @param scheduler the scheduler.
+	 */
+	public BatchingRabbitTemplate(BatchingStrategy batchingStrategy, TaskScheduler scheduler) {
+		this.batchingStrategy = batchingStrategy;
+		this.scheduler = scheduler;
+	}
+
+	@Override
+	public synchronized void send(String exchange, String routingKey, Message message) throws AmqpException {
+		if (this.scheduledTask != null) {
+			this.scheduledTask.cancel(false);
+		}
+		MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message);
+		if (batch != null) {
+			super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage());
+		}
+		Date next = this.batchingStrategy.nextRelease();
+		if (next != null) {
+			this.scheduledTask = this.scheduler.schedule(new Runnable() {
+
+				@Override
+				public void run() {
+					releaseBatches();
+				}}, next);
+		}
+	}
+
+	private synchronized void releaseBatches() {
+		MessageBatch batch;
+		while ((batch = this.batchingStrategy.releaseBatch()) != null) {
+			super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage());
+		}
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/BatchingStrategy.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.core.support;
+
+import java.util.Date;
+
+import org.springframework.amqp.core.Message;
+
+/**
+ * Strategy for batching messages. The methods will never be called concurrently.
+ * <p>
+ * <b>Experimental - APIs may change.</b>
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public interface BatchingStrategy {
+
+	/**
+	 * Add a message to the batch and optionally release the batch.
+	 * @param exchange The exchange.
+	 * @param routingKey The routing key.
+	 * @param message The message.
+	 * @return The batched message ({@link MessageBatch}), or null if not ready to release.
+	 */
+	MessageBatch addToBatch(String exchange, String routingKey, Message message);
+
+	/**
+	 * @return the date the next scheduled release should run, or null if no data to release.
+	 */
+	Date nextRelease();
+
+	/**
+	 * Release a batch, perhaps due to a timeout. May be called repeatedly
+	 * until {@code null} is returned.
+	 * @return The batched message, or null if no batches are ready.
+	 */
+	MessageBatch releaseBatch();
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/MessageBatch.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.core.support;
+
+import org.springframework.amqp.core.Message;
+
+/**
+ * An object encapsulating a {@link Message} containing the batch of messages,
+ * the exchange, and routing key.
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class MessageBatch {
+
+	private final String exchange;
+
+	private final String routingKey;
+
+	private final Message message;
+
+	public MessageBatch(String exchange, String routingKey, Message message) {
+		this.exchange = exchange;
+		this.routingKey = routingKey;
+		this.message = message;
+	}
+
+	/**
+	 * @return the exchange
+	 */
+	public String getExchange() {
+		return exchange;
+	}
+
+	/**
+	 * @return the routingKey
+	 */
+	public String getRoutingKey() {
+		return routingKey;
+	}
+
+	/**
+	 * @return the message
+	 */
+	public Message getMessage() {
+		return message;
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.core.support;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.util.Assert;
+
+/**
+ * A simple batching strategy that supports only one exchange/routingKey; includes a batch
+ * size, a batched message size limit and a timeout. The message properties from the first
+ * message in the batch is used in the batch message. Each message is preceded by a 4 byte
+ * length field.
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class SimpleBatchingStrategy implements BatchingStrategy {
+
+	private final int batchSize;
+
+	private final int bufferLimit;
+
+	private final long timeout;
+
+	private final List<Message> messages = new ArrayList<Message>();
+
+	private volatile String exchange;
+
+	private volatile String routingKey;
+
+	private volatile int currentSize;
+
+	/**
+	 * @param batchSize the batch size.
+	 * @param bufferLimit the max buffer size; could trigger a short batch. Does not apply
+	 * to a single message.
+	 * @param timeout the batch timeout.
+	 */
+	public SimpleBatchingStrategy(int batchSize, int bufferLimit, long timeout) {
+		this.batchSize = batchSize;
+		this.bufferLimit = bufferLimit;
+		this.timeout = timeout;
+	}
+
+	@Override
+	public MessageBatch addToBatch(String exchange, String routingKey, Message message) {
+		if (this.exchange != null) {
+			Assert.isTrue(this.exchange.equals(exchange), "Cannot send to different exchanges in the same batch");
+		}
+		else {
+			this.exchange = exchange;
+		}
+		if (this.routingKey != null) {
+			Assert.isTrue(this.routingKey.equals(routingKey), "Cannot send with different routing keys in the same batch");
+		}
+		else {
+			this.routingKey = routingKey;
+		}
+		int bufferUse = 4 + message.getBody().length;
+		MessageBatch batch = null;
+		if (this.messages.size() > 0 && this.currentSize + bufferUse > this.bufferLimit) {
+			batch = releaseBatch();
+			this.exchange = exchange;
+			this.routingKey = routingKey;
+		}
+		this.currentSize += bufferUse;
+		messages.add(message);
+		if (batch == null && (messages.size() >= this.batchSize
+								|| this.currentSize >= this.bufferLimit)) {
+			batch = releaseBatch();
+		}
+		return batch;
+	}
+
+	@Override
+	public Date nextRelease() {
+		if (this.messages.size() == 0 || this.timeout <= 0) {
+			return null;
+		}
+		else if (this.currentSize >= this.bufferLimit) {
+			// release immediately, we're already over the limit
+			return new Date();
+		}
+		else {
+			return new Date(System.currentTimeMillis() + this.timeout);
+		}
+	}
+
+	@Override
+	public MessageBatch releaseBatch() {
+		if (this.messages.size() < 1) {
+			return null;
+		}
+		Message message = assembleMessage();
+		MessageBatch messageBatch = new MessageBatch(this.exchange, this.routingKey, message);
+		this.messages.clear();
+		this.currentSize = 0;
+		this.exchange = null;
+		this.routingKey = null;
+		return messageBatch;
+	}
+
+	private Message assembleMessage() {
+		if (this.messages.size() == 1) {
+			return this.messages.get(0);
+		}
+		MessageProperties messageProperties = this.messages.get(0).getMessageProperties();
+		byte[] body = new byte[this.currentSize];
+		ByteBuffer bytes = ByteBuffer.wrap(body);
+		for (Message message : this.messages) {
+			bytes.putInt(message.getBody().length);
+			bytes.put(message.getBody());
+		}
+		messageProperties.getHeaders().put(MessageProperties.SPRING_BATCH_FORMAT, MessageProperties.BATCH_FORMAT_LENGTH_HEADER4);
+		return new Message(body, messageProperties);
+	}
+
+}

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

@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.listener;
 
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -26,6 +27,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.Connection;
@@ -37,6 +39,7 @@ import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
 import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
+import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
@@ -59,6 +62,8 @@ import com.rabbitmq.client.Channel;
 public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		implements MessageListenerContainer, ApplicationContextAware, BeanNameAware, DisposableBean, SmartLifecycle {
 
+	public static final boolean DEFAULT_DEBATCHING_ENABLED = true;
+
 	private volatile String beanName;
 
 	private volatile boolean autoStartup = true;
@@ -83,6 +88,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private volatile AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
 
+	private volatile boolean deBatchingEnabled = DEFAULT_DEBATCHING_ENABLED;
+
 	private boolean initialized;
 
 	private volatile ApplicationContext applicationContext;
@@ -298,6 +305,15 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		return messageConverter;
 	}
 
+	/**
+	 * Determine whether or not the container should de-batch batched
+	 * messages (true) or call the listener with the batch (false). Default: true.
+	 * @param deBatchingEnabled the deBatchingEnabled to set.
+	 */
+	protected void setDeBatchingEnabled(boolean deBatchingEnabled) {
+		this.deBatchingEnabled = deBatchingEnabled;
+	}
+
 	/**
 	 * Set whether to automatically start the container after initialization.
 	 * <p>
@@ -597,8 +613,31 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 			throw new MessageRejectedWhileStoppingException();
 		}
 		try {
-			invokeListener(channel, message);
-		} catch (Throwable ex) {
+			Object batchFormat = message.getMessageProperties().getHeaders().get(MessageProperties.SPRING_BATCH_FORMAT);
+			if (MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(batchFormat) && this.deBatchingEnabled) {
+				ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBody());
+				MessageProperties messageProperties = message.getMessageProperties();
+				messageProperties.getHeaders().remove(MessageProperties.SPRING_BATCH_FORMAT);
+				while (byteBuffer.hasRemaining()) {
+					int length = byteBuffer.getInt();
+					if (length < 0 || length > byteBuffer.remaining()) {
+						throw new ListenerExecutionFailedException("Bad batched message received",
+								new MessageConversionException("Insufficient batch data at offset " + byteBuffer.position()),
+										message);
+					}
+					byte[] body = new byte[length];
+					byteBuffer.get(body);
+					messageProperties.setContentLength(length);
+					// Caveat - shared MessageProperties.
+					Message fragment = new Message(body, messageProperties);
+					invokeListener(channel, fragment);
+				}
+			}
+			else {
+				invokeListener(channel, message);
+			}
+		}
+		catch (Throwable ex) {
 			handleListenerException(ex);
 			throw ex;
 		}

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

@@ -42,7 +42,7 @@ import org.springframework.util.ErrorHandler;
  */
 public class ConditionalRejectingErrorHandler implements ErrorHandler {
 
-	protected static final Log logger = LogFactory.getLog(ConditionalRejectingErrorHandler.class);
+	protected final Log logger = LogFactory.getLog(this.getClass());
 
 	private final FatalExceptionStrategy exceptionStrategy;
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java

@@ -0,0 +1,347 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.core;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.internal.stubbing.answers.DoesNothing;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
+import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
+import org.springframework.amqp.utils.test.TestUtils;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.StopWatch;
+
+/**
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class BatchingRabbitTemplateTests {
+
+	private static final String ROUTE = "test.queue";
+
+	@Rule
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE);
+
+	private CachingConnectionFactory connectionFactory;
+
+	private ThreadPoolTaskScheduler scheduler;
+
+	@Before
+	public void setup() {
+		this.connectionFactory = new CachingConnectionFactory();
+		this.connectionFactory.setHost("localhost");
+		this.connectionFactory.setPort(BrokerTestUtils.getPort());
+		scheduler = new ThreadPoolTaskScheduler();
+		scheduler.setPoolSize(1);
+		scheduler.initialize();
+	}
+
+	@Test
+	public void testSimpleBatch() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar", new String(message.getBody()));
+	}
+
+	@Test
+	public void testSimpleBatchTimeout() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 50);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("foo", new String(message.getBody()));
+	}
+
+	@Test
+	public void testSimpleBatchTimeoutMultiple() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 50);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003foo", new String(message.getBody()));
+	}
+
+	@Test
+	public void testSimpleBatchBufferLimit() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 8, 50);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("foo", new String(message.getBody()));
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("bar", new String(message.getBody()));
+	}
+
+	@Test
+	public void testSimpleBatchBufferLimitMultiple() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 15, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003foo", new String(message.getBody()));
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("\u0000\u0000\u0000\u0003bar\u0000\u0000\u0000\u0003bar", new String(message.getBody()));
+	}
+
+	@Test
+	public void testSimpleBatchBiggerThanBufferLimit() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 2, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("foo", new String(message.getBody()));
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("bar", new String(message.getBody()));
+	}
+
+	@Test
+	// existing buffered; new message bigger than bufferLimit; released immediately
+	public void testSimpleBatchBiggerThanBufferLimitMultiple() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 6, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("f".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("f", new String(message.getBody()));
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("bar", new String(message.getBody()));
+	}
+
+	@Test
+	public void testSimpleBatchTwoEqualBufferLimit() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(10, 14, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		Thread.sleep(100);
+		message = template.receive(ROUTE);
+		assertNotNull(message);
+		assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar", new String(message.getBody()));
+	}
+
+	@Test
+	public void testDebatchByContainer() throws Exception {
+		final List<Message> received = new ArrayList<Message>();
+		final CountDownLatch latch = new CountDownLatch(2);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
+		container.setQueueNames(ROUTE);
+		container.setMessageListener(new MessageListener() {
+
+			@Override
+			public void onMessage(Message message) {
+				received.add(message);
+				latch.countDown();
+			}
+		});
+		container.setReceiveTimeout(100);
+		container.afterPropertiesSet();
+		container.start();
+		try {
+			BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
+			BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+			template.setConnectionFactory(this.connectionFactory);
+			MessageProperties props = new MessageProperties();
+			Message message = new Message("foo".getBytes(), props);
+			template.send("", ROUTE, message);
+			message = new Message("bar".getBytes(), props);
+			template.send("", ROUTE, message);
+			assertTrue(latch.await(10,  TimeUnit.SECONDS));
+			assertEquals(2, received.size());
+			assertEquals("foo", new String(received.get(0).getBody()));
+			assertEquals(3, received.get(0).getMessageProperties().getContentLength());
+			assertEquals("bar", new String(received.get(1).getBody()));
+			assertEquals(3, received.get(0).getMessageProperties().getContentLength());
+		}
+		finally {
+			container.stop();
+		}
+	}
+
+	@Test
+	public void testDebatchByContainerPerformance() throws Exception {
+		final List<Message> received = new ArrayList<Message>();
+		int count = 100000;
+		final CountDownLatch latch = new CountDownLatch(count);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
+		container.setQueueNames(ROUTE);
+		container.setMessageListener(new MessageListener() {
+
+			@Override
+			public void onMessage(Message message) {
+				received.add(message);
+				latch.countDown();
+			}
+		});
+		container.setReceiveTimeout(100);
+		container.setPrefetchCount(1000);
+		container.setTxSize(1000);
+		container.afterPropertiesSet();
+		container.start();
+		try {
+			BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(1000, Integer.MAX_VALUE, 30000);
+			BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+//			RabbitTemplate template = new RabbitTemplate();
+			template.setConnectionFactory(this.connectionFactory);
+			MessageProperties props = new MessageProperties();
+			props.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+			Message message = new Message(new byte[256], props);
+			StopWatch watch = new StopWatch();
+			watch.start();
+			for (int i = 0; i < count; i++) {
+				template.send("", ROUTE, message);
+			}
+			assertTrue(latch.await(60,  TimeUnit.SECONDS));
+			watch.stop();
+			System.out.println(watch.getTotalTimeMillis());
+			assertEquals(count, received.size());
+		}
+		finally {
+			container.stop();
+		}
+	}
+
+	@Test
+	public void testDebatchByContainerBadMessageRejected() throws Exception {
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
+		container.setQueueNames(ROUTE);
+		container.setMessageListener(new MessageListener() {
+
+			@Override
+			public void onMessage(Message message) {
+			}
+		});
+		container.setReceiveTimeout(100);
+		ConditionalRejectingErrorHandler errorHandler = new ConditionalRejectingErrorHandler();
+		container.setErrorHandler(errorHandler);
+		container.afterPropertiesSet();
+		container.start();
+		Log logger = spy(TestUtils.getPropertyValue(errorHandler, "logger", Log.class));
+		new DirectFieldAccessor(errorHandler).setPropertyValue("logger", logger);
+		when(logger.isWarnEnabled()).thenReturn(true);
+		doAnswer(new DoesNothing()).when(logger).warn(anyString(), any(Throwable.class));
+		try {
+			RabbitTemplate template = new RabbitTemplate();
+			template.setConnectionFactory(this.connectionFactory);
+			MessageProperties props = new MessageProperties();
+			props.getHeaders().put(MessageProperties.SPRING_BATCH_FORMAT, MessageProperties.BATCH_FORMAT_LENGTH_HEADER4);
+			Message message = new Message("\u0000\u0000\u0000\u0004foo".getBytes(), props);
+			template.send("", ROUTE, message);
+			Thread.sleep(1000);
+			ArgumentCaptor<Object> arg1 = ArgumentCaptor.forClass(Object.class);
+			ArgumentCaptor<Throwable> arg2 = ArgumentCaptor.forClass(Throwable.class);
+			verify(logger, times(2)).warn(arg1.capture(), arg2.capture()); // CRE logs 2 WARNs ensure the message was rejected
+			assertThat(arg2.getValue().getMessage(), containsString("Bad batched message received"));
+		}
+		finally {
+			container.stop();
+		}
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/support/SimpleBatchStrategyTests.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.core.support;
+
+import java.nio.ByteBuffer;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import org.springframework.util.StopWatch;
+
+/**
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class SimpleBatchStrategyTests {
+
+	@Test @Ignore
+	public void testBatchingPerf() { // used to compare ByteBuffer Vs. System.arrayCopy()
+		StopWatch watch = new StopWatch();
+		byte[] bbBuff = new byte[10000];
+		ByteBuffer bb = ByteBuffer.wrap(bbBuff);
+		byte[] buff = new byte[10000];
+		watch.start();
+		for (int i = 0; i < 10000000; i++) {
+			bb.position(0);
+			bb.put(buff);
+//			System.arraycopy(buff, 0, bbBuff, 0, 10000);
+		}
+		watch.stop();
+		System.out.println(watch.getTotalTimeMillis());
+	}
+
+}