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

GH-265: Add ProducerMHCustomizer support

Browse Source 
Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/265

* Inject a possible `ProducerMessageHandlerCustomizer` in the `RabbitMessageChannelBinder`
* Ensure in the `RabbitBinderModuleTests.testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated()`
that customizer works
This commit is contained in:
Artem Bilan
2019-10-23 13:56:41 -04:00
committed by Gary Russell
parent 0028dedad5
commit 180cd9987e
2 changed files with 14 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -37,10 +37,12 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedB
import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.config.MessageSourceCustomizer;
import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.amqp.inbound.AmqpMessageSource;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.lang.Nullable;
/**
@@ -73,7 +75,8 @@ public class RabbitMessageChannelBinderConfiguration {
@Bean
RabbitMessageChannelBinder rabbitMessageChannelBinder(
@Nullable ListenerContainerCustomizer<AbstractMessageListenerContainer> listenerContainerCustomizer,
@Nullable MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer) {
@Nullable MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer,
@Nullable ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> producerMessageHandlerCustomizer) {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(
this.rabbitConnectionFactory, this.rabbitProperties,
@@ -84,6 +87,7 @@ public class RabbitMessageChannelBinderConfiguration {
binder.setDecompressingPostProcessor(deCompressingPostProcessor());
binder.setNodes(this.rabbitBinderConfigurationProperties.getNodes());
binder.setExtendedBindingProperties(this.rabbitExtendedBindingProperties);
binder.setProducerMessageHandlerCustomizer(producerMessageHandlerCustomizer);
return binder;
}

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

@@ -58,10 +58,12 @@ import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSuppo
import org.springframework.cloud.stream.binding.BindingService;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.config.MessageSourceCustomizer;
import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.amqp.inbound.AmqpMessageSource;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -179,6 +181,8 @@ public class RabbitBinderModuleTests {
Binding<MessageChannel> outputBinding = producerBindings.get("output");
assertThat(TestUtils.getPropertyValue(outputBinding,
"lifecycle.amqpTemplate.transactional", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(outputBinding, "lifecycle.beanName"))
.isEqualTo("setByCustomizer:output");
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
@@ -353,6 +357,11 @@ public class RabbitBinderModuleTests {
return (s, q, g) -> s.setBeanName("setByCustomizer:" + g);
}
@Bean
public ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> messageHandlerCustomizer() {
return (handler, destinationName) -> handler.setBeanName("setByCustomizer:" + destinationName);
}
}
public static class ConnectionFactoryConfiguration {