From 4fefd194640069addb49eb453e24e8a92f0b8fc7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 16 Jan 2019 13:43:04 -0500 Subject: [PATCH] GH-146: Support listener container type selection Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/146 --- docs/src/main/asciidoc/overview.adoc | 8 +++ .../properties/RabbitConsumerProperties.java | 14 +++++ .../rabbit/RabbitMessageChannelBinder.java | 56 ++++++++++++++----- .../binder/rabbit/RabbitBinderTests.java | 31 +++++++--- 4 files changed, 88 insertions(+), 21 deletions(-) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index e4ce35b30..14410e9cd 100644 --- 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`. diff --git 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 index 30d71550b..688f52cf1 100644 --- 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; + } + } diff --git 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 index fb310e2b9..80d41ed3e 100644 --- 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 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 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 consumerProperties) { diff --git 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 index 446042699..6d6553cfc 100644 --- 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 consumerProps = createConsumerProperties(); + consumerProps.getExtension().setContainerType(ContainerType.DIRECT); Binding 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 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 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 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 properties = createConsumerProperties(); @@ -992,6 +1006,7 @@ public class RabbitBinderTests extends outputBinding.unbind(); } + @SuppressWarnings("deprecation") @Test public void testAutoBindDLQPartitionedProducerFirst() throws Exception { RabbitTestBinder binder = getBinder();