INT-4402: Apply global interceptors at runtime

JIRA: https://jira.spring.io/browse/INT-4402

Make `GlobalChannelInterceptorProcessor` as a `BeanPostProcessor`,
so it can apply global `ChannelInterceptor`s to any initialized channel
beans, even those created at runtime, like in case of dynamic flows with
Java DSL

**Cherry-pick to 4.3.x**

* Add `postProcessDynamicBeans` global property with `false` by default
* Rely on the `postProcessDynamicBeans` property in the
`GlobalChannelInterceptorProcessor.postProcessAfterInitialization()`
*  Add `postProcessDynamicBeans=true` to the
`GlobalChannelInterceptorTests-context.xml` to be sure in the test-case
that option works
* Document the option and effect from the global channel interceptors

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-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.
@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Properties;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
@@ -33,18 +34,22 @@
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.core.OrderComparator;
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;
 import org.springframework.integration.channel.interceptor.VetoCapableInterceptor;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.util.PatternMatchUtils;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
- * Will apply global interceptors to channels (<channel-interceptor>).
+ * This class applies global interceptors ({@code <channel-interceptor>} or {@code @GlobalChannelInterceptor})
+ * to message channels beans.
  *
  * @author Oleg Zhurakousky
  * @author Mark Fisher
@@ -54,7 +59,8 @@
  *
  * @since 2.0
  */
-final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, SmartInitializingSingleton {
+public final class GlobalChannelInterceptorProcessor
+		implements BeanFactoryAware, SmartInitializingSingleton, BeanPostProcessor {
 
 	private static final Log logger = LogFactory.getLog(GlobalChannelInterceptorProcessor.class);
 
@@ -67,6 +73,8 @@ final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, Smart
 
 	private ListableBeanFactory beanFactory;
 
+	private volatile boolean singletonsInstantiated;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		Assert.isInstanceOf(ListableBeanFactory.class, beanFactory);
@@ -96,12 +104,29 @@ public void afterSingletonsInstantiated() {
 				addMatchingInterceptors(entry.getValue(), entry.getKey());
 			}
 		}
+
+		// TODO Remove this logic in 5.1
+		Properties integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
+
+		this.singletonsInstantiated =
+				Boolean.parseBoolean(integrationProperties.getProperty(
+						IntegrationProperties.POST_PROCESS_DYNAMIC_BEANS));
+	}
+
+	@Override
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		if (this.singletonsInstantiated && bean instanceof ChannelInterceptorAware) {
+			addMatchingInterceptors((ChannelInterceptorAware) bean, beanName);
+		}
+		return bean;
 	}
 
 	/**
-	 * Adds any interceptor whose pattern matches against the channel's name.
+	 * Add any interceptor whose pattern matches against the channel's name.
+	 * @param channel the message channel to add interceptors.
+	 * @param beanName the message channel bean name to match the pattern.
 	 */
-	private void addMatchingInterceptors(ChannelInterceptorAware channel, String beanName) {
+	public void addMatchingInterceptors(ChannelInterceptorAware channel, String beanName) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Applying global interceptors on channel '" + beanName + "'");
 		}

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-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.
@@ -77,6 +77,12 @@ public final class IntegrationProperties {
 	 */
 	public static final String ENDPOINTS_NO_AUTO_STARTUP = INTEGRATION_PROPERTIES_PREFIX + "endpoints.noAutoStartup";
 
+	/**
+	 * Whether {@link org.springframework.beans.factory.config.BeanPostProcessor}s should process beans registered at runtime.
+	 * Will be removed in 5.1.
+	 */
+	public static final String POST_PROCESS_DYNAMIC_BEANS = INTEGRATION_PROPERTIES_PREFIX + "postProcessDynamicBeans";
+
 
 	private static Properties defaults;
 

spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties

@@ -6,3 +6,4 @@ spring.integration.messagingTemplate.throwExceptionOnLateReply=false
 # Defaults to MessageHeaders.ID and MessageHeaders.TIMESTAMP
 spring.integration.readOnly.headers=
 spring.integration.endpoints.noAutoStartup=
+spring.integration.postProcessDynamicBeans=false

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests-context.xml

@@ -1,12 +1,18 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:p="http://www.springframework.org/schema/p"
-	xmlns:aop="http://www.springframework.org/schema/aop"
-	xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:p="http://www.springframework.org/schema/p"
+	   xmlns:aop="http://www.springframework.org/schema/aop" xmlns:util="http://www.springframework.org/schema/util"
+	   xmlns:beans="http://www.springframework.org/schema/c"
+	   xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
+
+	<util:properties id="integrationGlobalProperties">
+		<prop key="spring.integration.postProcessDynamicBeans">true</prop>
+	</util:properties>
 
 	<int:channel id="inputA">
 		<int:interceptors>

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-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,8 +16,10 @@
 
 package org.springframework.integration.channel.interceptor;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
@@ -31,9 +33,11 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.ApplicationContext;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.core.Ordered;
 import org.springframework.integration.channel.ChannelInterceptorAware;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
@@ -54,7 +58,7 @@
 public class GlobalChannelInterceptorTests {
 
 	@Autowired
-	ApplicationContext applicationContext;
+	ConfigurableApplicationContext applicationContext;
 
 	@Autowired
 	@Qualifier("inputC")
@@ -143,6 +147,18 @@ public void testWildCardPatternMatch() {
 		assertTrue(interceptorNames.contains("interceptor-eleven"));
 	}
 
+	@Test
+	public void testDynamicMessageChannelBeanWithAutoGlobalChannelInterceptor() {
+		DirectChannel testChannel = new DirectChannel();
+		ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory();
+		beanFactory.initializeBean(testChannel, "testChannel");
+
+		List<ChannelInterceptor> channelInterceptors = testChannel.getChannelInterceptors();
+
+		assertEquals(2, channelInterceptors.size());
+		assertThat(channelInterceptors.get(0), instanceOf(SampleInterceptor.class));
+		assertThat(channelInterceptors.get(0), instanceOf(SampleInterceptor.class));
+	}
 
 	public static class SampleInterceptor extends ChannelInterceptorAdapter {
 

src/reference/asciidoc/channel.adoc

@@ -734,6 +734,9 @@ To inject a global interceptor _BEFORE_ the existing interceptors, use a negativ
 NOTE: Note that both the `order` and `pattern` attributes are optional.
 The default value for `order` will be 0 and for `pattern`, the default is '*' (to match all channels).
 
+Starting with _version 4.3.15_, you can configure a property `spring.integration.postProcessDynamicBeans = true` to apply any global interceptors to dynamically created `MessageChannel` beans.
+See <<global-properties>> for more information.
+
 [[channel-wiretap]]
 ===== Wire Tap
 

src/reference/asciidoc/configuration.adoc

@@ -223,6 +223,7 @@ spring.integration.taskScheduler.poolSize=10 <4>
 spring.integration.messagingTemplate.throwExceptionOnLateReply=false <5>
 spring.integration.readOnly.headers= <6>
 spring.integration.endpoints.noAutoStartup= <7>
+spring.integration.postProcessDynamicBeans=false <8>
 ----
 
 <1> When true, `input-channel` s will be automatically declared as `DirectChannel` s when not explicitly found in the
@@ -245,11 +246,16 @@ expecting a reply - because the sending thread has timed out, or already receive
 The list is used by the `DefaultMessageBuilderFactory` bean and propagated to the `IntegrationMessageHeaderAccessor` instances (see <<message-header-accessor>>), used to build messages via `MessageBuilder` (see <<message-builder>>).
 By default only `MessageHeaders.ID` and `MessageHeaders.TIMESTAMP` are not copied during message building.
 _Since version 4.3.2_
+
 <7> A comma-separated list of `AbstractEndpoint` bean names patterns (`xxx*`, `*xxx`, `*xxx*` or `xxx*yyy`) which should not be started automatically during application startup.
 These endpoints can be started later manually by their bean name via `Control Bus` (see <<control-bus>>), by their role using the `SmartLifecycleRoleController` (see <<endpoint-roles>>) or via simple `Lifecycle` bean injection.
 The effect of this global property can be explicitly overridden by specifying `auto-startup` XML or `autoStartup` annotation attribute, or via call to the `AbstractEndpoint.setAutoStartup()` in bean definition.
 _Since version 4.3.12_
 
+<8> A boolean flag to indicate that `BeanPostProcessor` s should post-process beans registered at runtime, e.g. message channels created via `IntegrationFlowContext` can be supplied with global channel interceptors.
+_Since version 4.3.15_
+
+
 These properties can be overridden by adding a file `/META-INF/spring.integration.properties` to the classpath.
 It is not necessary to provide all the properties, just those that you want to override.
 
@@ -438,7 +444,7 @@ With this endpoint using the default poller:
 public class AnnotationService {
 
     @Transformer(inputChannel = "aPollableChannel", outputChannel = "output"
-                           poller = @Poller("myPoller")
+                           poller = @Poller("myPoller"))
     public String handle(String payload) {
          ...
     }
@@ -586,7 +592,7 @@ public class MyFlowConfiguration {
     public Supplier<String> pojoSupplier() {
         return () -> "foo";
     }
-	
+
     @Bean
     @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
     public Supplier<Message<String>> messageSupplier() {