GH-922: Add Batch-mode @RabbitListener

Resolves #922

- remove unnecessary back-ticks from literal table columns

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessagingMessageConverter.java

@@ -88,6 +88,10 @@ public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
 		this.headerMapper = headerMapper;
 	}
 
+	public AmqpHeaderMapper getHeaderMapper() {
+		return this.headerMapper;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		Assert.notNull(this.payloadConverter, "Property 'payloadConverter' is required");

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

@@ -29,6 +29,7 @@
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
@@ -119,6 +120,10 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private Consumer<C> containerConfigurer;
 
+	private boolean batchListener;
+
+	private BatchingStrategy batchingStrategy;
+
 	/**
 	 * @param connectionFactory The connection factory.
 	 * @see AbstractMessageListenerContainer#setConnectionFactory(ConnectionFactory)
@@ -344,6 +349,27 @@ public void setContainerConfigurer(Consumer<C> configurer) {
 		this.containerConfigurer = configurer;
 	}
 
+	/**
+	 * Set to true to receive a list of debatched messages that were created by a
+	 * {@link org.springframework.amqp.rabbit.core.BatchingRabbitTemplate}.
+	 * @param isBatch true for a batch listener.
+	 * @since 2.2
+	 * @see #setBatchingStrategy(BatchingStrategy)
+	 */
+	public void setBatchListener(boolean isBatch) {
+		this.batchListener = isBatch;
+	}
+
+	/**
+	 * Set a {@link BatchingStrategy} to use when debatching messages.
+	 * @param batchingStrategy the batching strategy.
+	 * @since 2.2
+	 * @see #setBatchListener(boolean)
+	 */
+	public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
+		this.batchingStrategy = batchingStrategy;
+	}
+
 	@SuppressWarnings("deprecation")
 	@Override
 	public C createListenerContainer(RabbitListenerEndpoint endpoint) {
@@ -381,12 +407,15 @@ public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 			.acceptIfNotNull(this.autoStartup, instance::setAutoStartup)
 			.acceptIfNotNull(this.phase, instance::setPhase)
 			.acceptIfNotNull(this.afterReceivePostProcessors, instance::setAfterReceivePostProcessors);
+		instance.setDeBatchingEnabled(!this.batchListener);
 		if (endpoint != null) { // endpoint settings overriding default factory settings
 			javaUtils
 				.acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup)
 				.acceptIfNotNull(endpoint.getTaskExecutor(), instance::setTaskExecutor);
+			javaUtils
+				.acceptIfNotNull(this.batchingStrategy, endpoint::setBatchingStrategy);
 			instance.setListenerId(endpoint.getId());
-
+			endpoint.setBatchListener(this.batchListener);
 			endpoint.setupListenerContainer(instance);
 		}
 		if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java

@@ -363,7 +363,7 @@ public static int getMaxFrame(ConnectionFactory connectionFactory) {
 				return rcon.getFrameMax();
 			}
 		}
-		catch (RuntimeException e) {
+		catch (@SuppressWarnings("unused") RuntimeException e) {
 			// NOSONAR
 		}
 		return -1;

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

@@ -21,9 +21,11 @@
 
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.core.support.MessageBatch;
+import org.springframework.lang.Nullable;
 import org.springframework.scheduling.TaskScheduler;
 
 /**
@@ -48,10 +50,25 @@ public class BatchingRabbitTemplate extends RabbitTemplate {
 	private volatile ScheduledFuture<?> scheduledTask;
 
 	/**
+	 * Create an instance with the supplied parameters.
 	 * @param batchingStrategy the batching strategy.
 	 * @param scheduler the scheduler.
 	 */
 	public BatchingRabbitTemplate(BatchingStrategy batchingStrategy, TaskScheduler scheduler) {
+		this(null, batchingStrategy, scheduler);
+	}
+
+	/**
+	 * Create an instance with the supplied parameters.
+	 * @param connectionFactory the connection factory.
+	 * @param batchingStrategy the batching strategy.
+	 * @param scheduler the scheduler.
+	 * @since 2.2
+	 */
+	public BatchingRabbitTemplate(@Nullable ConnectionFactory connectionFactory, BatchingStrategy batchingStrategy,
+			TaskScheduler scheduler) {
+
+		super(connectionFactory);
 		this.batchingStrategy = batchingStrategy;
 		this.scheduler = scheduler;
 	}

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

@@ -18,8 +18,10 @@
 
 import java.util.Collection;
 import java.util.Date;
+import java.util.function.Consumer;
 
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
 
 /**
  * Strategy for batching messages. The methods will never be called concurrently.
@@ -52,4 +54,27 @@ public interface BatchingStrategy {
 	 */
 	Collection<MessageBatch> releaseBatches();
 
+	/**
+	 * Return true if this strategy can decode a batch of messages from a message body.
+	 * Returning true means you must override {@link #deBatch(Message, Consumer)}.
+	 * @param properties the message properties.
+	 * @return true if we can decode the message.
+	 * @since 2.2
+	 * @see #deBatch(Message, Consumer)
+	 */
+	default boolean canDebatch(MessageProperties properties) {
+		return false;
+	}
+
+	/**
+	 * Decode a message into fragments.
+	 * @param message the message.
+	 * @param fragmentConsumer a consumer for fragments.
+	 * @since 2.2
+	 * @see #canDebatch(MessageProperties)
+	 */
+	default void deBatch(Message message, Consumer<Message> fragmentConsumer) {
+		throw new UnsupportedOperationException("Cannot debatch this message");
+	}
+
 }

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

@@ -22,9 +22,12 @@
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.function.Consumer;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
+import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.util.Assert;
 
 /**
@@ -66,25 +69,26 @@ public SimpleBatchingStrategy(int batchSize, int bufferLimit, long timeout) {
 	}
 
 	@Override
-	public MessageBatch addToBatch(String exchange, String routingKey, Message message) {
+	public MessageBatch addToBatch(String exch, String routKey, Message message) {
 		if (this.exchange != null) {
-			Assert.isTrue(this.exchange.equals(exchange), "Cannot send to different exchanges in the same batch");
+			Assert.isTrue(this.exchange.equals(exch), "Cannot send to different exchanges in the same batch");
 		}
 		else {
-			this.exchange = exchange;
+			this.exchange = exch;
 		}
 		if (this.routingKey != null) {
-			Assert.isTrue(this.routingKey.equals(routingKey), "Cannot send with different routing keys in the same batch");
+			Assert.isTrue(this.routingKey.equals(routKey),
+					"Cannot send with different routing keys in the same batch");
 		}
 		else {
-			this.routingKey = routingKey;
+			this.routingKey = routKey;
 		}
 		int bufferUse = Integer.BYTES + message.getBody().length;
 		MessageBatch batch = null;
 		if (this.messages.size() > 0 && this.currentSize + bufferUse > this.bufferLimit) {
 			batch = doReleaseBatch();
-			this.exchange = exchange;
-			this.routingKey = routingKey;
+			this.exchange = exch;
+			this.routingKey = routKey;
 		}
 		this.currentSize += bufferUse;
 		this.messages.add(message);
@@ -144,8 +148,45 @@ private Message assembleMessage() {
 			bytes.putInt(message.getBody().length);
 			bytes.put(message.getBody());
 		}
-		messageProperties.getHeaders().put(MessageProperties.SPRING_BATCH_FORMAT, MessageProperties.BATCH_FORMAT_LENGTH_HEADER4);
+		messageProperties.getHeaders().put(MessageProperties.SPRING_BATCH_FORMAT,
+				MessageProperties.BATCH_FORMAT_LENGTH_HEADER4);
 		return new Message(body, messageProperties);
 	}
 
+
+	@Override
+	public boolean canDebatch(MessageProperties properties) {
+		return MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(properties
+				.getHeaders()
+				.get(MessageProperties.SPRING_BATCH_FORMAT));
+	}
+
+	/**
+	 * Debatch a message that has a header with {@link MessageProperties#SPRING_BATCH_FORMAT}
+	 * set to {@link MessageProperties#BATCH_FORMAT_LENGTH_HEADER4}.
+	 * @param message the batched message.
+	 * @param fragmentConsumer a consumer for each fragment.
+	 * @since 2.2
+	 */
+	@Override
+	public void deBatch(Message message, Consumer<Message> fragmentConsumer) {
+		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);
+			fragmentConsumer.accept(fragment);
+		}
+	}
+
 }

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

@@ -16,7 +16,6 @@
 
 package org.springframework.amqp.rabbit.listener;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -44,7 +43,6 @@
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessagePostProcessor;
-import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -53,6 +51,8 @@
 import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.connection.RoutingConnectionFactory;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
 import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
@@ -61,7 +61,6 @@
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.ConditionalExceptionLogger;
 import org.springframework.amqp.support.ConsumerTagStrategy;
-import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.postprocessor.MessagePostProcessorUtils;
 import org.springframework.aop.framework.ProxyFactory;
@@ -222,6 +221,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private String errorHandlerLoggerName = getClass().getName();
 
+	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
+
 	private volatile boolean lazyLoad;
 
 	@Override
@@ -495,6 +496,7 @@ public MessageConverter getMessageConverter() {
 	 * 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.
+	 * @see #setBatchingStrategy(BatchingStrategy)
 	 */
 	public void setDeBatchingEnabled(boolean deBatchingEnabled) {
 		this.deBatchingEnabled = deBatchingEnabled;
@@ -1125,6 +1127,18 @@ public void setErrorHandlerLoggerName(String errorHandlerLoggerName) {
 		this.errorHandlerLoggerName = errorHandlerLoggerName;
 	}
 
+	/**
+	 * Set a batching strategy to use when de-batching messages.
+	 * Default is {@link SimpleBatchingStrategy}.
+	 * @param batchingStrategy the strategy.
+	 * @since 2.2
+	 * @see #setDeBatchingEnabled(boolean)
+	 */
+	public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
+		Assert.notNull(batchingStrategy, "'batchingStrategy' cannot be null");
+		this.batchingStrategy = batchingStrategy;
+	}
+
 	/**
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
@@ -1437,25 +1451,8 @@ private void doExecuteListener(Channel channel, Message messageIn) {
 				}
 			}
 		}
-		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);
-			}
+		if (this.deBatchingEnabled && this.batchingStrategy.canDebatch(message.getMessageProperties())) {
+			this.batchingStrategy.deBatch(message, fragment -> invokeListener(channel, fragment));
 		}
 		else {
 			invokeListener(channel, message);