GH-3204: Add DSL intercept() operator

Fixes https://github.com/spring-projects/spring-integration/issues/3204

* Add an `intercept(ChannelInterceptor...)` method into `BaseIntegrationFlowDefinition` 
to register one or more channel interceptors at the current flow position.
* refactor to reuse `InterceptableChannel` creation from `wireTap`
* document the new operator

spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java

@@ -90,6 +90,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.InterceptableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -109,6 +110,7 @@ import reactor.util.function.Tuple2;
  * @author Artem Bilan
  * @author Gary Russell
  * @author Gabriele Del Prete
+ * @author Tim Feuerbach
  *
  * @since 5.2.1
  *
@@ -179,6 +181,24 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return this.currentMessageChannel;
 	}
 
+	/**
+	 * Return the current channel if it is an {@link InterceptableChannel}, otherwise register a new implicit
+	 * {@link DirectChannel} in the flow and return that one.
+	 * @return the current channel after the operation
+	 */
+	protected InterceptableChannel currentInterceptableChannel() {
+		MessageChannel currentChannel = getCurrentMessageChannel();
+		if (currentChannel instanceof InterceptableChannel) {
+			return (InterceptableChannel) currentChannel;
+		}
+		else {
+			DirectChannel newCurrentChannel = new DirectChannel();
+			channel(newCurrentChannel);
+			setImplicitChannel(true);
+			return newCurrentChannel;
+		}
+	}
+
 	protected void setImplicitChannel(boolean implicitChannel) {
 		this.implicitChannel = implicitChannel;
 	}
@@ -488,14 +508,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 */
 	public B wireTap(WireTapSpec wireTapSpec) {
 		WireTap interceptor = wireTapSpec.get();
-		MessageChannel currentChannel = getCurrentMessageChannel();
-		if (!(currentChannel instanceof InterceptableChannel)) {
-			currentChannel = new DirectChannel();
-			channel(currentChannel);
-			setImplicitChannel(true);
-		}
+		InterceptableChannel currentChannel = currentInterceptableChannel();
 		addComponent(wireTapSpec);
-		((InterceptableChannel) currentChannel).addInterceptor(interceptor);
+		currentChannel.addInterceptor(interceptor);
 		return _this();
 	}
 
@@ -2829,6 +2844,26 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return handle(new ServiceActivatingHandler(triggerAction, "trigger"), endpointConfigurer);
 	}
 
+	/**
+	 * Add one or more {@link ChannelInterceptor} implementations
+	 * to the current {@link #currentMessageChannel}, in the given order, after any interceptors already registered.
+	 * @param interceptorArray one or more {@link ChannelInterceptor}s.
+	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @throws IllegalArgumentException if one or more null arguments are provided
+	 * @since 5.3
+	 */
+	public B intercept(ChannelInterceptor... interceptorArray) {
+		Assert.notNull(interceptorArray, "'interceptorArray' must not be null");
+		Assert.noNullElements(interceptorArray, "'interceptorArray' must not contain null elements");
+
+		InterceptableChannel currentChannel = currentInterceptableChannel();
+		for (ChannelInterceptor interceptor : interceptorArray) {
+			currentChannel.addInterceptor(interceptor);
+		}
+
+		return _this();
+	}
+
 	/**
 	 * Populate a {@link FluxMessageChannel} to start a reactive processing for upstream data,
 	 * wrap it to a {@link Flux}, apply provided {@link Function} via {@link Flux#transform(Function)}

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -20,6 +20,8 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
@@ -32,6 +34,7 @@ import java.util.function.Supplier;
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
+import org.junit.After;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.BeanCreationException;
@@ -83,6 +86,7 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -100,6 +104,7 @@ import reactor.core.publisher.Mono;
  * @author Tim Ysewyn
  * @author Gary Russell
  * @author Oleg Zhurakousky
+ * @author Tim Feuerbach
  *
  * @since 5.0
  */
@@ -498,6 +503,32 @@ public class IntegrationFlowTests {
 		this.errorChannel.unsubscribe(errorMessageHandler);
 	}
 
+	@Autowired
+	@Qualifier("interceptorChannelIn")
+	private MessageChannel interceptorChannelIn;
+
+	@Autowired
+	private List<String> outputStringList;
+
+	@Test
+	public void testInterceptorFlow() {
+		this.interceptorChannelIn.send(MessageBuilder.withPayload("foo").build());
+
+		assertThat(outputStringList).containsExactly(
+				"Pre send transform: foo",
+				"Pre send handle: FOO",
+				"Handle: FOO",
+				"Post send handle: FOO",
+				"Post send transform: foo"
+		);
+	}
+
+	@After
+	public void cleanUpList() {
+		outputStringList.clear();
+	}
+
+
 	@MessagingGateway
 	public interface ControlBusGateway {
 
@@ -909,6 +940,46 @@ public class IntegrationFlowTests {
 
 	}
 
+	@Configuration
+	public static class InterceptorContextConfiguration {
+
+		@Bean
+		public List<String> outputStringList() {
+			return new ArrayList<>();
+		}
+
+		@Bean
+		public IntegrationFlow interceptorFlow(List<String> outputStringList) {
+			return IntegrationFlows.from("interceptorChannelIn")
+					.intercept(new ChannelInterceptor() {
+						@Override
+						public Message<?> preSend(Message<?> message, MessageChannel channel) {
+							outputStringList.add("Pre send transform: " + message.getPayload());
+							return message;
+						}
+
+						@Override
+						public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+							outputStringList.add("Post send transform: " + message.getPayload());
+						}
+					})
+					.transform((String s) -> s.toUpperCase())
+					.intercept(new ChannelInterceptor() {
+						@Override
+						public Message<?> preSend(Message<?> message, MessageChannel channel) {
+							outputStringList.add("Pre send handle: " + message.getPayload());
+							return message;
+						}
+
+						@Override
+						public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+							outputStringList.add("Post send handle: " + message.getPayload());
+						}
+					})
+					.handle(m -> outputStringList.add("Handle: " + m.getPayload())).get();
+		}
+	}
+
 	@Service
 	public static class GreetingService extends AbstractReplyProducingMessageHandler {