GH-146: Support listener container type selection

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/146
2019-01-16 13:43:04 -05:00
parent 5dca68feac
commit 8f12671da0
4 changed files with 88 additions and 21 deletions
--- a/docs/src/main/asciidoc/overview.adoc
+++ b/docs/src/main/asciidoc/overview.adoc
@@ -139,6 +139,11 @@ Used to create the consumer tag(s); will be appended by `#n` where `n` increment
 Example: `${spring.application.name}-${spring.cloud.stream.bindings.input.group}-${spring.cloud.stream.instance-index}`.
 +
 Default: none - the broker will generate random consumer tags.
+containerType::
+Select the type of listener container to be used.
+See https://docs.spring.io/spring-amqp/reference/html/_reference.html#choose-container[Choosing a Container] in the Spring AMQP documentation for more information.
+
+Default: `simple`
 deadLetterQueueName::
 The name of the DLQ
 +
@@ -264,6 +269,7 @@ Consider using a policy instead of this setting, because using a policy allows c
 Default: `false`.
 maxConcurrency::
 The maximum number of consumers.
+Not supported when the `containerType` is `direct`.
 +
 Default: `1`.
 maxLength::
@@ -299,6 +305,7 @@ queueDeclarationRetries::
 The number of times to retry consuming from a queue if it is missing.
 Relevant only when `missingQueuesFatal` is `true`.
 Otherwise, the container keeps retrying indefinitely.
+Not supported when the `containerType` is `direct`.
 +
 Default: `3`
 queueNameGroupOnly::
@@ -337,6 +344,7 @@ Default time to live to apply to the queue when declared (in milliseconds).
 Default: `no limit`
 txSize::
 The number of deliveries between acks.
+Not supported when the `containerType` is `direct`.
 +
 Default: `1`.

--- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java
+++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java
@@ -20,6 +20,7 @@ import javax.validation.constraints.Min;

 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType;
 import org.springframework.util.Assert;

 /**
@@ -114,6 +115,11 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	 */
 	private int frameMaxHeadroom = 20_000;

+	/**
+	 * The container type, SIMPLE or DIRECT.
+	 */
+	private ContainerType containerType = ContainerType.SIMPLE;
+
 	public boolean isTransacted() {
 		return transacted;
 	}
@@ -272,4 +278,12 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 		this.frameMaxHeadroom = frameMaxHeadroom;
 	}

+	public ContainerType getContainerType() {
+		return this.containerType;
+	}
+
+	public void setContainerType(ContainerType containerType) {
+		this.containerType = containerType;
+	}
+
 }
--- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
+++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2018 the original author or authors.
+ * Copyright 2013-2019 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.
@@ -38,6 +38,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
@@ -51,6 +52,7 @@ import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPos
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.BinderHeaders;
@@ -70,11 +72,11 @@ import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.integration.acks.AcknowledgmentCallback.Status;
-import org.springframework.expression.Expression;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.inbound.AmqpMessageSource;
 import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
@@ -389,21 +391,23 @@ public class RabbitMessageChannelBinder
 		Assert.state(!HeaderMode.embeddedHeaders.equals(properties.getHeaderMode()),
 				"the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively");
 		String destination = consumerDestination.getName();
-		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
-				this.connectionFactory);
+		boolean directContainer = properties.getExtension().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());
 		listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
 		int concurrency = properties.getConcurrency();
 		concurrency = concurrency > 0 ? concurrency : 1;
-		listenerContainer.setConcurrentConsumers(concurrency);
-		int maxConcurrency = properties.getExtension().getMaxConcurrency();
-		if (maxConcurrency > concurrency) {
-			listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
+		if (directContainer) {
+			setDMLCProperties(properties, (DirectMessageListenerContainer) listenerContainer, concurrency);
+		}
+		else {
+			setSMLCProperties(properties, (SimpleMessageListenerContainer) listenerContainer, concurrency);
 		}
 		listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
 		listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
-		listenerContainer.setTxSize(properties.getExtension().getTxSize());
 		listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
 		String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true);
 		listenerContainer.setQueueNames(queues);
@@ -412,9 +416,6 @@ public class RabbitMessageChannelBinder
 				RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
 		listenerContainer.setExclusive(properties.getExtension().isExclusive());
 		listenerContainer.setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal());
-		if (properties.getExtension().getQueueDeclarationRetries() != null) {
-			listenerContainer.setDeclarationRetries(properties.getExtension().getQueueDeclarationRetries());
-		}
 		if (properties.getExtension().getFailedDeclarationRetryInterval() != null) {
 			listenerContainer.setFailedDeclarationRetryInterval(
 					properties.getExtension().getFailedDeclarationRetryInterval());
@@ -452,6 +453,35 @@ public class RabbitMessageChannelBinder
 		return adapter;
 	}

+	private void setSMLCProperties(ExtendedConsumerProperties<RabbitConsumerProperties> properties,
+			SimpleMessageListenerContainer listenerContainer, int concurrency) {
+
+		listenerContainer.setConcurrentConsumers(concurrency);
+		int maxConcurrency = properties.getExtension().getMaxConcurrency();
+		if (maxConcurrency > concurrency) {
+			listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
+		}
+		listenerContainer.setTxSize(properties.getExtension().getTxSize());
+		if (properties.getExtension().getQueueDeclarationRetries() != null) {
+			listenerContainer.setDeclarationRetries(properties.getExtension().getQueueDeclarationRetries());
+		}
+	}
+
+	private void setDMLCProperties(ExtendedConsumerProperties<RabbitConsumerProperties> properties,
+			DirectMessageListenerContainer listenerContainer, int concurrency) {
+
+		listenerContainer.setConsumersPerQueue(concurrency);
+		if (properties.getExtension().getMaxConcurrency() > concurrency) {
+			this.logger.warn("maxConcurrency is not supported with a direct container type");
+		}
+		if (properties.getExtension().getTxSize() > 1) {
+			this.logger.warn("txSize is not supported with a direct container type");
+		}
+		if (properties.getExtension().getQueueDeclarationRetries() != null) {
+			this.logger.warn("queueDeclarationRetries is not supported with a direct container type");
+		}
+	}
+
 	@Override
 	protected PolledConsumerResources createPolledConsumerResources(String name, String group, ConsumerDestination destination,
 			ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
--- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
+++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2018 the original author or authors.
+ * Copyright 2013-2019 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.
@@ -51,10 +51,10 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.amqp.rabbit.core.RabbitManagementTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.AsyncConsumerStartedEvent;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
 import org.springframework.amqp.support.AmqpHeaders;
@@ -63,6 +63,7 @@ import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
@@ -186,10 +187,14 @@ public class RabbitBinderTests extends
 			.isTrue();
 		assertThat(TestUtils.getPropertyValue(producerBinding, "lifecycle.amqpTemplate.messageConverter")
 				.getClass().getName()).contains("Passthrough");
+		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProps = createConsumerProperties();
+		consumerProps.getExtension().setContainerType(ContainerType.DIRECT);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel,
-				createConsumerProperties());
-		assertThat(TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageConverter")
-				.getClass().getName()).contains("Passthrough");
+				consumerProps);
+		assertThat(TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageConverter").getClass().getName())
+				.contains("Passthrough");
+		assertThat(TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageListenerContainer"))
+				.isInstanceOf(DirectMessageListenerContainer.class);
 		Message<?> message = MessageBuilder.withPayload("bad".getBytes()).setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
 		final CountDownLatch latch = new CountDownLatch(3);
 		moduleInputChannel.subscribe(new MessageHandler() {
@@ -363,6 +368,7 @@ public class RabbitBinderTests extends
 		assertThat(endpoint.isRunning()).isFalse();
 	}

+	@SuppressWarnings("deprecation")
 	@Test
 	public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception {
 		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
@@ -386,7 +392,8 @@ public class RabbitBinderTests extends
 		assertThat(container.isRunning()).isTrue();
 		consumerBinding.unbind();
 		assertThat(container.isRunning()).isFalse();
-		RabbitManagementTemplate rmt = new RabbitManagementTemplate();
+		org.springframework.amqp.rabbit.core.RabbitManagementTemplate rmt =
+				new org.springframework.amqp.rabbit.core.RabbitManagementTemplate();
 		List<org.springframework.amqp.core.Binding> bindings = rmt.getBindingsForExchange("/", exchange.getName());
 		assertThat(bindings.size()).isEqualTo(1);
 	}
@@ -410,6 +417,7 @@ public class RabbitBinderTests extends
 		assertThat(container.isRunning()).isFalse();
 	}

+	@SuppressWarnings("deprecation")
 	@Test
 	public void testConsumerPropertiesWithUserInfrastructureCustomExchangeAndRK() throws Exception {
 		RabbitTestBinder binder = getBinder();
@@ -429,7 +437,8 @@ public class RabbitBinderTests extends
 		consumerBinding.unbind();
 		assertThat(container.isRunning()).isFalse();
 		assertThat(container.getQueueNames()[0]).isEqualTo(group);
-		RabbitManagementTemplate rmt = new RabbitManagementTemplate();
+		org.springframework.amqp.rabbit.core.RabbitManagementTemplate rmt =
+				new org.springframework.amqp.rabbit.core.RabbitManagementTemplate();
 		List<org.springframework.amqp.core.Binding> bindings = rmt.getBindingsForExchange("/", "propsUser2");
 		int n = 0;
 		while (n++ < 100 && bindings == null || bindings.size() < 1) {
@@ -457,6 +466,7 @@ public class RabbitBinderTests extends
 		assertThat(exchange.isAutoDelete()).isEqualTo(false);
 	}

+	@SuppressWarnings("deprecation")
 	@Test
 	public void testConsumerPropertiesWithUserInfrastructureCustomQueueArgs() throws Exception {
 		RabbitTestBinder binder = getBinder();
@@ -496,7 +506,8 @@ public class RabbitBinderTests extends
 		SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
 				SimpleMessageListenerContainer.class);
 		assertThat(container.isRunning()).isTrue();
-		RabbitManagementTemplate rmt = new RabbitManagementTemplate();
+		org.springframework.amqp.rabbit.core.RabbitManagementTemplate rmt =
+				new org.springframework.amqp.rabbit.core.RabbitManagementTemplate();
 		List<org.springframework.amqp.core.Binding> bindings = rmt.getBindingsForExchange("/", "propsUser3");
 		int n = 0;
 		while (n++ < 100 && bindings == null || bindings.size() < 1) {
@@ -570,6 +581,7 @@ public class RabbitBinderTests extends
 		assertThat(container.isRunning()).isFalse();
 	}

+	@SuppressWarnings("deprecation")
 	@Test
 	public void testProducerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();
@@ -765,6 +777,7 @@ public class RabbitBinderTests extends
 		assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq")).isFalse();
 	}

+	@SuppressWarnings("deprecation")
 	@Test
 	public void testAutoBindDLQPartionedConsumerFirst() throws Exception {
 		RabbitTestBinder binder = getBinder();
@@ -866,6 +879,7 @@ public class RabbitBinderTests extends
 		testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(true);
 	}

+	@SuppressWarnings("deprecation")
 	private void testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(final boolean withRetry) throws Exception {
 		RabbitTestBinder binder = getBinder();
 		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
@@ -992,6 +1006,7 @@ public class RabbitBinderTests extends
 		outputBinding.unbind();
 	}

+	@SuppressWarnings("deprecation")
 	@Test
 	public void testAutoBindDLQPartitionedProducerFirst() throws Exception {
 		RabbitTestBinder binder = getBinder();