GH-1219 Fixed channel instrumentation in BinderAwareChannelResolver

- added support for instrumenting dynamic channels with global interceptors
- added test

Resolves #1219
Resolves #1227

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2017 the original author or authors.
+ * Copyright 2013-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.
@@ -16,11 +16,16 @@
 
 package org.springframework.cloud.stream.binding;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.integration.channel.ChannelInterceptorAware;
+import org.springframework.integration.config.GlobalChannelInterceptorProcessor;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver;
 import org.springframework.messaging.core.DestinationResolutionException;
@@ -35,8 +40,11 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Ilayaperumal Gopinathan
+ * @author Oleg Zhurakousky
  */
 public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestinationResolver {
+	
+	private final Log logger = LogFactory.getLog(BinderAwareChannelResolver.class);
 
 	private final BindingService bindingService;
 
@@ -48,23 +56,34 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 	private final NewDestinationBindingCallback newBindingCallback;
 
 	private ConfigurableListableBeanFactory beanFactory;
-
+	
+	private final GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor;
+	
 	public BinderAwareChannelResolver(BindingService bindingService,
 			AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory,
 			DynamicDestinationsBindable dynamicDestinationsBindable) {
-		this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null);
+		this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null, null);
+	}
+	
+	@SuppressWarnings("rawtypes")
+	public BinderAwareChannelResolver(BindingService bindingService,
+			AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory,
+			DynamicDestinationsBindable dynamicDestinationsBindable, NewDestinationBindingCallback callback) {
+		this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, callback, null);
 	}
 
 	@SuppressWarnings("rawtypes")
 	public BinderAwareChannelResolver(BindingService bindingService,
 			AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory,
-			DynamicDestinationsBindable dynamicDestinationsBindable, NewDestinationBindingCallback callback) {
+			DynamicDestinationsBindable dynamicDestinationsBindable, NewDestinationBindingCallback callback, 
+			GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) {
 		this.dynamicDestinationsBindable = dynamicDestinationsBindable;
 		Assert.notNull(bindingService, "'bindingService' cannot be null");
 		Assert.notNull(bindingTargetFactory, "'bindingTargetFactory' cannot be null");
 		this.bindingService = bindingService;
 		this.bindingTargetFactory = bindingTargetFactory;
 		this.newBindingCallback = callback;
+		this.globalChannelInterceptorProcessor = globalChannelInterceptorProcessor;
 	}
 
 	@Override
@@ -95,8 +114,7 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 			MessageChannel channel = null;
 			if (this.beanFactory != null) {
 				String[] dynamicDestinations = null;
-				BindingServiceProperties bindingServiceProperties = this.bindingService
-						.getBindingServiceProperties();
+				BindingServiceProperties bindingServiceProperties = this.bindingService.getBindingServiceProperties();
 				if (bindingServiceProperties != null) {
 					dynamicDestinations = bindingServiceProperties.getDynamicDestinations();
 				}
@@ -105,6 +123,9 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 				if (dynamicAllowed) {
 					channel = this.bindingTargetFactory.createOutput(channelName);
 					this.beanFactory.registerSingleton(channelName, channel);
+					
+					this.instrumentChannelWithGlobalInterceptors(channel, channelName);
+							
 					channel = (MessageChannel) this.beanFactory.initializeBean(channel, channelName);
 					if (this.newBindingCallback != null) {
 						ProducerProperties producerProperties = this.bindingService.getBindingServiceProperties()
@@ -126,6 +147,17 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 			return channel;
 		}
 	}
+	
+	private void instrumentChannelWithGlobalInterceptors(MessageChannel channel, String channelName) {
+		if (channel instanceof ChannelInterceptorAware) {
+			if (this.globalChannelInterceptorProcessor != null) {
+				this.globalChannelInterceptorProcessor.addMatchingInterceptors((ChannelInterceptorAware) channel, channelName);
+			}				
+		}
+		else {
+			logger.warn("Failed to add global interceptors to '" + channelName + "' since it is not an instance of ChannelInterceptorAware.");
+		}
+	}
 
 	/**
 	 * Configure a new destination before it is bound.
@@ -151,5 +183,4 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 				T extendedProducerProperties);
 
 	}
-
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java

@@ -55,6 +55,7 @@ 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.GlobalChannelInterceptorProcessor;
 import org.springframework.integration.config.HandlerMethodArgumentResolversHolder;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -175,9 +176,10 @@ public class BindingServiceConfiguration {
 	public BinderAwareChannelResolver binderAwareChannelResolver(BindingService bindingService,
 			AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory,
 			DynamicDestinationsBindable dynamicDestinationsBindable,
-			@Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback) {
+			@Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback, 
+			@Nullable GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) {
 		return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable,
-				callback);
+				callback, globalChannelInterceptorProcessor);
 	}
 
 	@Bean

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

@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.Before;
 import org.junit.Test;
+
 import org.mockito.Mockito;
 
 import org.springframework.boot.WebApplicationType;
@@ -38,16 +39,22 @@ import org.springframework.cloud.stream.binding.SubscribableChannelBindingTarget
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ImmutableMessageChannelInterceptor;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
@@ -76,10 +83,20 @@ public class BinderAwareChannelResolverTests {
 
 	protected volatile DynamicDestinationsBindable dynamicDestinationsBindable;
 
+	@Configuration
+	public static class InterceptorConfiguration {
+		@Bean
+		public GlobalChannelInterceptorWrapper testInterceptor() {
+			return  new GlobalChannelInterceptorWrapper(new ImmutableMessageChannelInterceptor());
+		}
+	}
+	
 	@SuppressWarnings("unchecked")
 	@Before
 	public void setupContext() throws Exception {
-		this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()).web(WebApplicationType.NONE).run();
+		this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration(BinderAwareChannelResolverTests.InterceptorConfiguration.class))
+				.web(WebApplicationType.NONE).run();
+		
 		this.resolver = context.getBean(BinderAwareChannelResolver.class);
 		this.binder = context.getBean(Binder.class);
 		this.bindingServiceProperties = context.getBean(BindingServiceProperties.class);
@@ -95,6 +112,8 @@ public class BinderAwareChannelResolverTests {
 			assertEquals(0, bindable.getOutputs().size());// consumer
 		}
 		MessageChannel registered = resolver.resolveDestination("foo");
+		assertEquals(2, ((AbstractMessageChannel)registered).getChannelInterceptors().size());
+		assertTrue(((AbstractMessageChannel)registered).getChannelInterceptors().get(1) instanceof ImmutableMessageChannelInterceptor);
 
 		bindables = context.getBeansOfType(Bindable.class);
 		assertThat(bindables).hasSize(1);