SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-139: Advanced listener container configuration

Browse Source 
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.
This commit is contained in:
Gary Russell
2018-04-16 17:54:53 -04:00
committed by Oleg Zhurakousky
parent 95c2f7ee49
commit f7f801d0ee
5 changed files with 74 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc
View File

@@ -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

11
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -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);

43
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java Normal file
View File

@@ -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);
}

9
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java
View File

@@ -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());

10
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -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<MessageChannel> 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<String, Binding<MessageChannel>> producerBindings = (Map<String, Binding<MessageChannel>>) 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 {