GH-1210 Fixed AbstractReplyProducingMessageHandler post processing

- addNotPropagatedHeaders(..) is now moved to simple ApplicationListener listening for ContextRefreshedEvent

Resolves #1210

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<AbstractReplyProducingMessageHandler> producingMessageHandlers,
-			@Autowired SpringIntegrationProperties springIntegrationProperties) {
-		return new InitializingBean() {
+	public ApplicationListener<ContextRefreshedEvent> appListener(SpringIntegrationProperties springIntegrationProperties) {
+		return new ApplicationListener<ContextRefreshedEvent>() {
 			@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()));
 			}
 		};
 	}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java

@@ -59,12 +59,15 @@ public class SpringIntegrationBinderConfiguration<T> {
 	 * application annotated with {@link EnableBinding}, yet require
 	 * full {@link Binder} configuration.
 	 */
-	public static Class<?>[] getCompleteConfiguration() {
+	public static Class<?>[] getCompleteConfiguration(Class<?>... additionalConfigurationClasses) {
 		List<Class<?>> configClasses = new ArrayList<>();
 		configClasses.add(SpringIntegrationBinderConfiguration.class);
 		Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, Import.class);
 		Map<String, Object> 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<?>[] {});
 	}
 

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<String, AbstractReplyProducingMessageHandler> 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) {
+		}
+	}
+	
+	
+}