From 0332d16c489658a3677ffb39cc174a84e4accbf9 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 8 Feb 2018 17:36:50 -0600 Subject: [PATCH] GH-1210 Fixed AbstractReplyProducingMessageHandler post processing - addNotPropagatedHeaders(..) is now moved to simple ApplicationListener listening for ContextRefreshedEvent Resolves #1210 --- .../config/BindingServiceConfiguration.java | 19 ++---- .../SpringIntegrationBinderConfiguration.java | 5 +- .../BindingServiceConfigurationTests.java | 66 +++++++++++++++++++ 3 files changed, 77 insertions(+), 13 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 80a928b8e..7a1d6a7c8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -20,7 +20,6 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; -import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanDefinition; @@ -47,11 +46,13 @@ import org.springframework.cloud.stream.binding.SingleBindingTargetBindable; import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.DependsOn; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Role; +import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.config.HandlerMethodArgumentResolversHolder; @@ -66,7 +67,6 @@ import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.scheduling.TaskScheduler; -import org.springframework.util.CollectionUtils; /** * Configuration class that provides necessary beans for {@link MessageChannel} binding. @@ -229,17 +229,12 @@ public class BindingServiceConfiguration { } @Bean - public static InitializingBean messageHandlerHeaderPropagationBeanPostProcessor(@Autowired(required=false) List producingMessageHandlers, - @Autowired SpringIntegrationProperties springIntegrationProperties) { - return new InitializingBean() { + public ApplicationListener appListener(SpringIntegrationProperties springIntegrationProperties) { + return new ApplicationListener() { @Override - public void afterPropertiesSet() throws Exception { - if (!CollectionUtils.isEmpty(producingMessageHandlers)) { - String[] messageHandlerNotPropagatedHeaders = springIntegrationProperties.getMessageHandlerNotPropagatedHeaders(); - for (AbstractReplyProducingMessageHandler producingMessageHandler : producingMessageHandlers) { - producingMessageHandler.addNotPropagatedHeaders(messageHandlerNotPropagatedHeaders); - } - } + public void onApplicationEvent(ContextRefreshedEvent event) { + event.getApplicationContext().getBeansOfType(AbstractReplyProducingMessageHandler.class).values() + .forEach(mh -> mh.addNotPropagatedHeaders(springIntegrationProperties.getMessageHandlerNotPropagatedHeaders())); } }; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java index 56cb544ec..98e24b638 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java @@ -59,12 +59,15 @@ public class SpringIntegrationBinderConfiguration { * application annotated with {@link EnableBinding}, yet require * full {@link Binder} configuration. */ - public static Class[] getCompleteConfiguration() { + public static Class[] getCompleteConfiguration(Class... additionalConfigurationClasses) { List> configClasses = new ArrayList<>(); configClasses.add(SpringIntegrationBinderConfiguration.class); Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, Import.class); Map annotationAttributes = AnnotationUtils.getAnnotationAttributes(annotation); configClasses.addAll(Arrays.asList((Class[])annotationAttributes.get("value"))); + if (additionalConfigurationClasses != null) { + configClasses.addAll(Arrays.asList(additionalConfigurationClasses)); + } return configClasses.toArray(new Class[] {}); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java new file mode 100644 index 000000000..986418eca --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java @@ -0,0 +1,66 @@ +/* + * Copyright 2017 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.config; + +import java.util.Map; + +import org.junit.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; + +import static org.junit.Assert.assertTrue; + +/** + * + * @author Oleg Zhurakousky + * + */ +public class BindingServiceConfigurationTests { + + @Test + public void valdateImportedConfiguartionHandlerPostProcessing() { + ApplicationContext context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration(RootConfiguration.class)).web(WebApplicationType.NONE).run(); + Map beansOfType = context.getBeansOfType(AbstractReplyProducingMessageHandler.class); + for (AbstractReplyProducingMessageHandler handler : beansOfType.values()) { + assertTrue(handler.getNotPropagatedHeaders().contains("contentType")); + } + } + + @Configuration + @Import(ImportedConfiguration.class) + public static class RootConfiguration { + @ServiceActivator(inputChannel="input") + public void rootService(String val) { + } + } + + @Configuration + public static class ImportedConfiguration { + @ServiceActivator(inputChannel="input") + public void importedService(String val) { + } + } + + +}