Spring AMQP: Support Both Container Types

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2012-2017 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.boot.autoconfigure.amqp;
+
+import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
+import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ListenerRetry;
+import org.springframework.util.Assert;
+
+/**
+ * Configure {@link RabbitListenerContainerFactory} with sensible defaults.
+ *
+ * @param <T> the container factory type.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public abstract class AbstractRabbitListenerContainerFactoryConfigurer<
+			T extends AbstractRabbitListenerContainerFactory<?>> {
+
+	private MessageConverter messageConverter;
+
+	private MessageRecoverer messageRecoverer;
+
+	private RabbitProperties rabbitProperties;
+
+	/**
+	 * Set the {@link MessageConverter} to use or {@code null} if the out-of-the-box
+	 * converter should be used.
+	 * @param messageConverter the {@link MessageConverter}
+	 */
+	protected void setMessageConverter(MessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+	}
+
+	/**
+	 * Set the {@link MessageRecoverer} to use or {@code null} to rely on the default.
+	 * @param messageRecoverer the {@link MessageRecoverer}
+	 */
+	protected void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		this.messageRecoverer = messageRecoverer;
+	}
+
+	/**
+	 * Set the {@link RabbitProperties} to use.
+	 * @param rabbitProperties the {@link RabbitProperties}
+	 */
+	protected void setRabbitProperties(RabbitProperties rabbitProperties) {
+		this.rabbitProperties = rabbitProperties;
+	}
+
+	/**
+	 * Configure the specified rabbit listener container factory. The factory can be
+	 * further tuned and default settings can be overridden.
+	 * @param factory the {@link AbstractRabbitListenerContainerFactory} instance to
+	 * configure
+	 * @param connectionFactory the {@link ConnectionFactory} to use
+	 */
+	public final void configure(T factory, ConnectionFactory connectionFactory) {
+		Assert.notNull(factory, "Factory must not be null");
+		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
+		factory.setConnectionFactory(connectionFactory);
+		if (this.messageConverter != null) {
+			factory.setMessageConverter(this.messageConverter);
+		}
+		RabbitProperties.Listener listenerConfig = this.rabbitProperties.getListener();
+		factory.setAutoStartup(listenerConfig.isAutoStartup());
+		if (listenerConfig.getAcknowledgeMode() != null) {
+			factory.setAcknowledgeMode(listenerConfig.getAcknowledgeMode());
+		}
+		if (listenerConfig.getPrefetch() != null) {
+			factory.setPrefetchCount(listenerConfig.getPrefetch());
+		}
+		if (listenerConfig.getDefaultRequeueRejected() != null) {
+			factory.setDefaultRequeueRejected(listenerConfig.getDefaultRequeueRejected());
+		}
+		if (listenerConfig.getIdleEventInterval() != null) {
+			factory.setIdleEventInterval(listenerConfig.getIdleEventInterval());
+		}
+		ListenerRetry retryConfig = listenerConfig.getRetry();
+		if (retryConfig.isEnabled()) {
+			RetryInterceptorBuilder<?> builder = (retryConfig.isStateless()
+					? RetryInterceptorBuilder.stateless()
+					: RetryInterceptorBuilder.stateful());
+			builder.maxAttempts(retryConfig.getMaxAttempts());
+			builder.backOffOptions(retryConfig.getInitialInterval(),
+					retryConfig.getMultiplier(), retryConfig.getMaxInterval());
+			MessageRecoverer recoverer = (this.messageRecoverer != null
+					? this.messageRecoverer : new RejectAndDontRequeueRecoverer());
+			builder.recoverer(recoverer);
+			factory.setAdviceChain(builder.build());
+		}
+		configure(factory, this.rabbitProperties);
+	}
+
+	/**
+	 * Perform factory-specific configuration.
+	 *
+	 * @param factory the factory.
+	 * @param rabbitProperties the properties.
+	 */
+	protected abstract void configure(T factory, RabbitProperties rabbitProperties);
+
+}

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/DirectRabbitListenerContainerFactoryConfigurer.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2012-2017 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.boot.autoconfigure.amqp;
+
+import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
+
+/**
+ * Configure {@link RabbitListenerContainerFactory} with sensible defaults.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ */
+public final class DirectRabbitListenerContainerFactoryConfigurer
+		extends AbstractRabbitListenerContainerFactoryConfigurer<DirectRabbitListenerContainerFactory> {
+
+	@Override
+	protected void configure(DirectRabbitListenerContainerFactory factory, RabbitProperties rabbitProperties) {
+		RabbitProperties.Listener listenerConfig = rabbitProperties.getListener();
+		if (listenerConfig.getConsumersPerQueue() != null) {
+			factory.setConsumersPerQueue(listenerConfig.getConsumersPerQueue());
+		}
+	}
+
+}

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java

@@ -17,6 +17,7 @@
 package org.springframework.boot.autoconfigure.amqp;
 
 import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -25,6 +26,7 @@
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -39,11 +41,11 @@
 @ConditionalOnClass(EnableRabbit.class)
 class RabbitAnnotationDrivenConfiguration {
 
-	private final ObjectProvider<MessageConverter> messageConverter;
+	protected final ObjectProvider<MessageConverter> messageConverter;
 
-	private final ObjectProvider<MessageRecoverer> messageRecoverer;
+	protected final ObjectProvider<MessageRecoverer> messageRecoverer;
 
-	private final RabbitProperties properties;
+	protected final RabbitProperties properties;
 
 	RabbitAnnotationDrivenConfiguration(ObjectProvider<MessageConverter> messageConverter,
 			ObjectProvider<MessageRecoverer> messageRecoverer,
@@ -53,24 +55,68 @@ class RabbitAnnotationDrivenConfiguration {
 		this.properties = properties;
 	}
 
-	@Bean
-	@ConditionalOnMissingBean
-	public SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() {
-		SimpleRabbitListenerContainerFactoryConfigurer configurer = new SimpleRabbitListenerContainerFactoryConfigurer();
-		configurer.setMessageConverter(this.messageConverter.getIfUnique());
-		configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique());
-		configurer.setRabbitProperties(this.properties);
-		return configurer;
+
+	@Configuration
+	@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "listener.type", havingValue = "simple",
+							matchIfMissing = true)
+	public static class SimpleContainerConfiguration extends RabbitAnnotationDrivenConfiguration {
+
+		SimpleContainerConfiguration(ObjectProvider<MessageConverter> messageConverter,
+				ObjectProvider<MessageRecoverer> messageRecoverer, RabbitProperties properties) {
+			super(messageConverter, messageRecoverer, properties);
+		}
+
+		@Bean
+		@ConditionalOnMissingBean
+		public SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() {
+			SimpleRabbitListenerContainerFactoryConfigurer configurer =
+					new SimpleRabbitListenerContainerFactoryConfigurer();
+			configurer.setMessageConverter(this.messageConverter.getIfUnique());
+			configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique());
+			configurer.setRabbitProperties(this.properties);
+			return configurer;
+		}
+
+		@Bean
+		@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
+		public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
+				SimpleRabbitListenerContainerFactoryConfigurer configurer,
+				ConnectionFactory connectionFactory) {
+			SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+			configurer.configure(factory, connectionFactory);
+			return factory;
+		}
 	}
 
-	@Bean
-	@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
-	public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
-			SimpleRabbitListenerContainerFactoryConfigurer configurer,
-			ConnectionFactory connectionFactory) {
-		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
-		configurer.configure(factory, connectionFactory);
-		return factory;
+	@Configuration
+	@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "listener.type", havingValue = "direct")
+	public static class DirectContainerConfiguration extends RabbitAnnotationDrivenConfiguration {
+
+		DirectContainerConfiguration(ObjectProvider<MessageConverter> messageConverter,
+				ObjectProvider<MessageRecoverer> messageRecoverer, RabbitProperties properties) {
+			super(messageConverter, messageRecoverer, properties);
+		}
+
+		@Bean
+		@ConditionalOnMissingBean
+		public DirectRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() {
+			DirectRabbitListenerContainerFactoryConfigurer configurer =
+					new DirectRabbitListenerContainerFactoryConfigurer();
+			configurer.setMessageConverter(this.messageConverter.getIfUnique());
+			configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique());
+			configurer.setRabbitProperties(this.properties);
+			return configurer;
+		}
+
+		@Bean
+		@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
+		public DirectRabbitListenerContainerFactory rabbitListenerContainerFactory(
+				DirectRabbitListenerContainerFactoryConfigurer configurer,
+				ConnectionFactory connectionFactory) {
+			DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
+			configurer.configure(factory, connectionFactory);
+			return factory;
+		}
 	}
 
 	@EnableRabbit

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java

@@ -39,6 +39,22 @@
 @ConfigurationProperties(prefix = "spring.rabbitmq")
 public class RabbitProperties {
 
+	public enum ContainerType {
+
+		/**
+		 * SimpleMessageListenerContainer - legacy container where the RabbitMQ consumer
+		 * dispatches messages to an invoker thread.
+		 */
+		SIMPLE,
+
+		/**
+		 * DirectMessageListenerContainer - container where the listener is invoked
+		 * directly on the RabbitMQ consumer thread.
+		 */
+		DIRECT
+
+	}
+
 	/**
 	 * RabbitMQ host.
 	 */
@@ -466,6 +482,11 @@ public void setSize(Integer size) {
 
 	public static class Listener {
 
+		/**
+		 * Container type.
+		 */
+		private ContainerType type = ContainerType.SIMPLE;
+
 		/**
 		 * Start the container automatically on startup.
 		 */
@@ -477,24 +498,30 @@ public static class Listener {
 		private AcknowledgeMode acknowledgeMode;
 
 		/**
-		 * Minimum number of consumers.
+		 * Minimum number of listener invoker threads - applies only to simple containers.
 		 */
 		private Integer concurrency;
 
 		/**
-		 * Maximum number of consumers.
+		 * Maximum number of listener invoker threads - applies only to simple containers.
 		 */
 		private Integer maxConcurrency;
 
+		/**
+		 * Number of RabbitMQ consumers per queue - applies only to direct containers.
+		 */
+		private Integer consumersPerQueue;
+
 		/**
 		 * Number of messages to be handled in a single request. It should be greater than
 		 * or equal to the transaction size (if used).
 		 */
 		private Integer prefetch;
 
 		/**
-		 * Number of messages to be processed in a transaction. For best results it should
-		 * be less than or equal to the prefetch count.
+		 * Number of messages to be processed in a transaction; number of messages between
+		 * acks. For best results it should be less than or equal to the prefetch count -
+		 * applies only to simple containers.
 		 */
 		private Integer transactionSize;
 
@@ -514,6 +541,14 @@ public static class Listener {
 		@NestedConfigurationProperty
 		private final ListenerRetry retry = new ListenerRetry();
 
+		public ContainerType getType() {
+			return this.type;
+		}
+
+		public void setType(ContainerType containerType) {
+			this.type = containerType;
+		}
+
 		public boolean isAutoStartup() {
 			return this.autoStartup;
 		}
@@ -546,6 +581,14 @@ public void setMaxConcurrency(Integer maxConcurrency) {
 			this.maxConcurrency = maxConcurrency;
 		}
 
+		public Integer getConsumersPerQueue() {
+			return this.consumersPerQueue;
+		}
+
+		public void setConsumersPerQueue(Integer consumersPerQueue) {
+			this.consumersPerQueue = consumersPerQueue;
+		}
+
 		public Integer getPrefetch() {
 			return this.prefetch;
 		}