From f7f801d0ee2e16c404810485830495b26a2640cd Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 16 Apr 2018 17:54:53 -0400 Subject: [PATCH] GH-139: Advanced listener container configuration Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/139 Allow configuration of listener container properties that are not exposed as binding or binder properties. --- .../src/main/asciidoc/overview.adoc | 6 +++ .../rabbit/RabbitMessageChannelBinder.java | 11 +++++ .../config/ListenerContainerCustomizer.java | 43 +++++++++++++++++++ ...bbitMessageChannelBinderConfiguration.java | 9 ++-- .../integration/RabbitBinderModuleTests.java | 10 +++++ 5 files changed, 74 insertions(+), 5 deletions(-) create mode 100644 spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index 0de58b736..2442eb016 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -311,6 +311,12 @@ The number of deliveries between acks. + Default: `1`. +=== Advanced Listener Container Configuration + +To set listener container properties that are not exposed as binder or binding properties, add a single bean of type `ListenerContainerCustomizer` to the application context. +The binder and binding properties will be set and then the customizer will be called. +The customizer (`configure()` method) is provided with the queue name as well as the consumer group as arguments. + === Rabbit Producer Properties The following properties are available for Rabbit producers only and 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 2bceaa61d..b508849a7 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 @@ -55,6 +55,7 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.HeaderMode; +import org.springframework.cloud.stream.binder.rabbit.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; @@ -131,6 +132,8 @@ public class RabbitMessageChannelBinder private final RabbitProperties rabbitProperties; + private final ListenerContainerCustomizer containerCustomizer; + private boolean destroyConnectionFactory; private ConnectionFactory connectionFactory; @@ -149,11 +152,18 @@ public class RabbitMessageChannelBinder public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, RabbitExchangeQueueProvisioner provisioningProvider) { + this(connectionFactory, rabbitProperties, provisioningProvider, null); + } + + public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, + RabbitExchangeQueueProvisioner provisioningProvider, + ListenerContainerCustomizer containerCustomizer) { super(new String[0], provisioningProvider); Assert.notNull(connectionFactory, "connectionFactory must not be null"); Assert.notNull(rabbitProperties, "rabbitProperties must not be null"); this.connectionFactory = connectionFactory; this.rabbitProperties = rabbitProperties; + this.containerCustomizer = containerCustomizer == null ? (c, q, g) -> { } : containerCustomizer; } /** @@ -389,6 +399,7 @@ public class RabbitMessageChannelBinder else if (getApplicationContext() != null) { listenerContainer.setApplicationEventPublisher(getApplicationContext()); } + this.containerCustomizer.configure(listenerContainer, consumerDestination.getName(), group); listenerContainer.afterPropertiesSet(); AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java new file mode 100644 index 000000000..abedb9f16 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java @@ -0,0 +1,43 @@ +/* + * Copyright 2018 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.cloud.stream.binder.rabbit.config; + +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; + +/** + * If a single bean of this type is in the application context, listener + * containers created by the binder can be further customized after all + * the properties are set. For example, to configure less-common + * properties. + * + * @author Gary Russell + * @since 2.1 + * + */ +@FunctionalInterface +public interface ListenerContainerCustomizer { + + /** + * Configure the container that is being created for the supplied queue name and + * consumer group. + * @param container the container. + * @param queueName the destination name. + * @param group the consumer group. + */ + void configure(AbstractMessageListenerContainer container, String queueName, String group); + +} diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index e7cc6bafa..432920978 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -34,7 +34,6 @@ import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; -import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -53,9 +52,6 @@ import org.springframework.context.annotation.Import; @EnableConfigurationProperties({ RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class }) public class RabbitMessageChannelBinderConfiguration { - @Autowired - private ConfigurableApplicationContext applicationContext; - @Autowired private ConnectionFactory rabbitConnectionFactory; @@ -68,10 +64,13 @@ public class RabbitMessageChannelBinderConfiguration { @Autowired private RabbitExtendedBindingProperties rabbitExtendedBindingProperties; + @Autowired(required = false) + private ListenerContainerCustomizer containerCustomizer; + @Bean RabbitMessageChannelBinder rabbitMessageChannelBinder() throws Exception { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory, - this.rabbitProperties, provisioningProvider()); + this.rabbitProperties, provisioningProvider(), this.containerCustomizer); binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); binder.setDecompressingPostProcessor(deCompressingPostProcessor()); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 961ae418e..b62a60941 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -47,6 +47,7 @@ import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.rabbit.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; @@ -140,6 +141,7 @@ public class RabbitBinderModuleTests { context = new SpringApplicationBuilder(SimpleProcessor.class) .web(WebApplicationType.NONE) .run("--server.port=0", + "--spring.cloud.stream.bindings.input.group=someGroup", "--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true", "--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true"); BinderFactory binderFactory = context.getBean(BinderFactory.class); @@ -152,6 +154,8 @@ public class RabbitBinderModuleTests { Binding inputBinding = consumerBindings.get("input").get(0); SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding, "lifecycle.messageListenerContainer", SimpleMessageListenerContainer.class); + assertThat(TestUtils.getPropertyValue(container, "beanName")) + .isEqualTo("setByCustomizerForQueue:input.someGroup,andGroup:someGroup"); assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue(); Map> producerBindings = (Map>) TestUtils .getPropertyValue(bindingService, "producerBindings"); @@ -286,6 +290,12 @@ public class RabbitBinderModuleTests { @SpringBootApplication public static class SimpleProcessor { + @Bean + public ListenerContainerCustomizer containerCustomizer() { + return (c, q, g) -> c.setBeanName("setByCustomizerForQueue:" + q + + (g == null ? "" : ",andGroup:" + g)); + } + } public static class ConnectionFactoryConfiguration {