diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index 42e7d81c4..a2a249d87 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java +++ b/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 bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable) { - this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null); + this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null, null); + } + + @SuppressWarnings("rawtypes") + public BinderAwareChannelResolver(BindingService bindingService, + AbstractBindingTargetFactory bindingTargetFactory, + DynamicDestinationsBindable dynamicDestinationsBindable, NewDestinationBindingCallback callback) { + this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, callback, null); } @SuppressWarnings("rawtypes") public BinderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory 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); } - } 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 7a1d6a7c8..a165a4325 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 @@ -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 bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable, - @Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback) { + @Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback, + @Nullable GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) { return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable, - callback); + callback, globalChannelInterceptorProcessor); } @Bean diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 2fb09db7c..5ec4e02a4 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/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);