INT-4434: Allow to use sub-flows from beans

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

There were a restriction introduced since Java DSL `1.2` do not use
`IntegrationFlow` beans for sub-flow definitions, e.g. in routers.
It is considered as regression by community because it worked before
in version `1.1`

* Introduce `IntegrationFlow.getInputChannel()` to be able to bridge
from the main flow to the flow which is treated as sub-flow.
In most cases we talk about an independent bean for the `IntegrationFlow`
which can be used as a stand along one and as a sub-flow in other flow

**Cherry-pick to 5.0.x**

Add JavaDocs to the `EndpointSpec.obtainInputChannelFromFlow()`

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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,10 +16,8 @@
 
 package org.springframework.integration.dsl;
 
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.router.AbstractMessageRouter;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.util.Assert;
 
 /**
  * A {@link MessageHandlerSpec} for {@link AbstractMessageRouter}s.
@@ -64,7 +62,6 @@ public S applySequence(boolean applySequence) {
 	 * Specify a {@link MessageChannel} bean name as a default output from the router.
 	 * @param channelName the {@link MessageChannel} bean name.
 	 * @return the router spec.
-	 * @since 1.2
 	 * @see AbstractMessageRouter#setDefaultOutputChannelName(String)
 	 */
 	public S defaultOutputChannel(String channelName) {
@@ -76,7 +73,6 @@ public S defaultOutputChannel(String channelName) {
 	 * Specify a {@link MessageChannel} as a default output from the router.
 	 * @param channel the {@link MessageChannel} to use.
 	 * @return the router spec.
-	 * @since 1.2
 	 * @see AbstractMessageRouter#setDefaultOutputChannel(MessageChannel)
 	 */
 	public S defaultOutputChannel(MessageChannel channel) {
@@ -88,25 +84,16 @@ public S defaultOutputChannel(MessageChannel channel) {
 	 * Specify an {@link IntegrationFlow} as an output from the router when no any other mapping has matched.
 	 * @param subFlow the {@link IntegrationFlow} for default mapping.
 	 * @return the router spec.
-	 * @since 1.2
 	 */
 	public S defaultSubFlowMapping(IntegrationFlow subFlow) {
-		Assert.notNull(subFlow, "'subFlow' must not be null");
-		DirectChannel channel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
-		subFlow.configure(flowBuilder);
-
-		this.componentsToRegister.put(flowBuilder, null);
-
-		return defaultOutputChannel(channel);
+		return defaultOutputChannel(obtainInputChannelFromFlow(subFlow, false));
 	}
 
 	/**
 	 * Make a default output mapping of the router to the parent flow.
 	 * Use the next, after router, parent flow {@link MessageChannel} as a
 	 * {@link AbstractMessageRouter#setDefaultOutputChannel(MessageChannel)} of this router.
 	 * @return the router spec.
-	 * @since 1.2
 	 */
 	public S defaultOutputToParentFlow() {
 		this.defaultToParentFlow = true;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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,8 +23,10 @@
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.ResolvableType;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.endpoint.AbstractPollingEndpoint;
 import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 
@@ -141,4 +143,42 @@ protected void assertHandler() {
 		Assert.state(this.handler != null, "'this.handler' must not be null.");
 	}
 
+	/**
+	 * Try to get a {@link MessageChannel} as an input for the provided {@link IntegrationFlow}
+	 * or create one and wrap the provided flow to a new one.
+	 * @param subFlow the {@link IntegrationFlow} to extract input channel.
+	 * @return the input channel of the flow of create one
+	 * @since 5.0.4
+	 */
+	protected MessageChannel obtainInputChannelFromFlow(IntegrationFlow subFlow) {
+		return obtainInputChannelFromFlow(subFlow, true);
+	}
+
+	/**
+	 * Try to get a {@link MessageChannel} as an input for the provided {@link IntegrationFlow}
+	 * or create one and wrap the provided flow to a new one.
+	 * @param subFlow the {@link IntegrationFlow} to extract input channel.
+	 * @param evaluateInternalBuilder true if an internal {@link IntegrationFlowDefinition} should be
+	 * evaluated to an {@link IntegrationFlow} component or left as a builder in the {@link #componentsToRegister}
+	 * for future use-case. For example the builder is used for router configurations to retain beans
+	 * registration order for parent-child dependencies.
+	 * @return the input channel of the flow of create one
+	 * @since 5.0.4
+	 */
+	protected MessageChannel obtainInputChannelFromFlow(IntegrationFlow subFlow, boolean evaluateInternalBuilder) {
+		Assert.notNull(subFlow, "'subFlow' must not be null");
+		MessageChannel messageChannel = subFlow.getInputChannel();
+		if (messageChannel == null) {
+			messageChannel = new DirectChannel();
+			IntegrationFlowDefinition<?> flowBuilder = IntegrationFlows.from(messageChannel);
+			subFlow.configure(flowBuilder);
+			this.componentsToRegister.put(evaluateInternalBuilder ? flowBuilder.get() : flowBuilder, null);
+		}
+		else {
+			this.componentsToRegister.put(subFlow, null);
+		}
+
+		return messageChannel;
+	}
+
 }

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

@@ -21,7 +21,6 @@
 import java.util.function.Function;
 
 import org.springframework.expression.Expression;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.expression.ValueExpression;
@@ -48,10 +47,9 @@
  */
 public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnricher> {
 
-	private final Map<String, Expression> propertyExpressions = new HashMap<String, Expression>();
+	private final Map<String, Expression> propertyExpressions = new HashMap<>();
 
-	private final Map<String, HeaderValueMessageProcessor<?>> headerExpressions =
-			new HashMap<String, HeaderValueMessageProcessor<?>>();
+	private final Map<String, HeaderValueMessageProcessor<?>> headerExpressions = new HashMap<>();
 
 	EnricherSpec() {
 		super(new ContentEnricher());
@@ -167,15 +165,7 @@ public <P> EnricherSpec requestPayload(Function<Message<P>, ?> requestPayloadFun
 	 * @return the enricher spec
 	 */
 	public EnricherSpec requestSubFlow(IntegrationFlow subFlow) {
-		Assert.notNull(subFlow, "'subFlow' must not be null");
-
-		DirectChannel requestChannel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(requestChannel);
-		subFlow.configure(flowBuilder);
-
-		this.componentsToRegister.put(flowBuilder.get(), null);
-
-		return requestChannel(requestChannel);
+		return requestChannel(obtainInputChannelFromFlow(subFlow));
 	}
 
 	/**
@@ -196,7 +186,7 @@ public EnricherSpec shouldClonePayload(boolean shouldClonePayload) {
 	 * @see ContentEnricher#setPropertyExpressions(Map)
 	 */
 	public <V> EnricherSpec property(String key, V value) {
-		this.propertyExpressions.put(key, new ValueExpression<V>(value));
+		this.propertyExpressions.put(key, new ValueExpression<>(value));
 		return _this();
 	}
 
@@ -234,7 +224,7 @@ public <P> EnricherSpec propertyFunction(String key, Function<Message<P>, Object
 	 * @see ContentEnricher#setHeaderExpressions(Map)
 	 */
 	public <V> EnricherSpec header(String name, V value) {
-		return this.header(name, value, null);
+		return header(name, value, null);
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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,10 +16,8 @@
 
 package org.springframework.integration.dsl;
 
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.filter.MessageFilter;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.util.Assert;
 
 /**
  * A {@link ConsumerEndpointSpec} implementation for the {@link MessageFilter}.
@@ -87,12 +85,7 @@ public FilterEndpointSpec discardChannel(String discardChannelName) {
 	 * @return the endpoint spec.
 	 */
 	public FilterEndpointSpec discardFlow(IntegrationFlow discardFlow) {
-		Assert.notNull(discardFlow, "'discardFlow' must not be null");
-		DirectChannel channel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
-		discardFlow.configure(flowBuilder);
-		this.componentsToRegister.put(flowBuilder.get(), null);
-		return discardChannel(channel);
+		return discardChannel(obtainInputChannelFromFlow(discardFlow));
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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,6 +16,8 @@
 
 package org.springframework.integration.dsl;
 
+import org.springframework.messaging.MessageChannel;
+
 /**
  * The main Integration DSL abstraction.
  * <p>
@@ -80,4 +82,14 @@ public interface IntegrationFlow {
 	 */
 	void configure(IntegrationFlowDefinition<?> flow);
 
+	/**
+	 * Return the first {@link MessageChannel} component
+	 * which is essential a flow input channel.
+	 * @return the channel.
+	 * @since 5.0.4
+	 */
+	default MessageChannel getInputChannel() {
+		return null;
+	}
+
 }

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

@@ -68,6 +68,12 @@ public final void configure(IntegrationFlowDefinition<?> flow) {
 		this.targetIntegrationFlow = flow.get();
 	}
 
+	@Override
+	public MessageChannel getInputChannel() {
+		assertTargetIntegrationFlow();
+		return this.targetIntegrationFlow.getInputChannel();
+	}
+
 	@Override
 	public void start() {
 		assertTargetIntegrationFlow();

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

@@ -74,11 +74,6 @@ public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
 	}
 
-	@Override
-	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
-		return bean;
-	}
-
 	@Override
 	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
 		if (bean instanceof StandardIntegrationFlow) {

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

@@ -334,13 +334,27 @@ public B wireTap(MessageChannel wireTapChannel) {
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public B wireTap(IntegrationFlow flow, Consumer<WireTapSpec> wireTapConfigurer) {
-		DirectChannel wireTapChannel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(wireTapChannel);
-		flow.configure(flowBuilder);
-		addComponent(flowBuilder.get());
+		MessageChannel wireTapChannel = obtainInputChannelFromFlow(flow);
+
 		return wireTap(wireTapChannel, wireTapConfigurer);
 	}
 
+	private MessageChannel obtainInputChannelFromFlow(IntegrationFlow flow) {
+		Assert.notNull(flow, "'flow' must not be null");
+		MessageChannel messageChannel = flow.getInputChannel();
+		if (messageChannel == null) {
+			messageChannel = new DirectChannel();
+			IntegrationFlowDefinition<?> flowBuilder = IntegrationFlows.from(messageChannel);
+			flow.configure(flowBuilder);
+			addComponent(flowBuilder.get());
+		}
+		else {
+			addComponent(flow);
+		}
+
+		return messageChannel;
+	}
+
 	/**
 	 * Populate the {@code Wire Tap} EI Pattern specific
 	 * {@link org.springframework.messaging.support.ChannelInterceptor} implementation
@@ -2164,11 +2178,8 @@ public B gateway(IntegrationFlow flow) {
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public B gateway(IntegrationFlow flow, Consumer<GatewayEndpointSpec> endpointConfigurer) {
-		Assert.notNull(flow, "'flow' must not be null");
-		final DirectChannel requestChannel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(requestChannel);
-		flow.configure(flowBuilder);
-		addComponent(flowBuilder.get());
+		MessageChannel requestChannel = obtainInputChannelFromFlow(flow);
+
 		return gateway(requestChannel, endpointConfigurer);
 	}
 
@@ -2700,9 +2711,9 @@ protected StandardIntegrationFlow get() {
 		if (this.integrationFlow == null) {
 			if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) {
 				throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel
-						+ ") is a prototype for FixedSubscriberChannel which can't be created without MessageHandler "
+						+ ") is a prototype for 'FixedSubscriberChannel' which can't be created without 'MessageHandler' "
 						+ "constructor argument. That means that '.fixedSubscriberChannel()' can't be the last "
-						+ "EIP-method in the IntegrationFlow definition.");
+						+ "EIP-method in the 'IntegrationFlow' definition.");
 			}
 
 			if (this.integrationComponents.size() == 1) {

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

@@ -20,6 +20,9 @@
 import java.util.Map;
 import java.util.concurrent.Executor;
 
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
 /**
  * @author Artem Bilan
  *
@@ -42,11 +45,21 @@ public PublishSubscribeSpec id(String id) {
 		return super.id(id);
 	}
 
-	public PublishSubscribeSpec subscribe(IntegrationFlow flow) {
+	public PublishSubscribeSpec subscribe(IntegrationFlow subFlow) {
+		Assert.notNull(subFlow, "'subFlow' must not be null");
+
 		IntegrationFlowBuilder flowBuilder =
 				IntegrationFlows.from(this.channel)
 						.bridge();
-		flow.configure(flowBuilder);
+
+		MessageChannel subFlowInput = subFlow.getInputChannel();
+
+		if (subFlowInput == null) {
+			subFlow.configure(flowBuilder);
+		}
+		else {
+			flowBuilder.channel(subFlowInput);
+		}
 		this.subscriberFlows.put(flowBuilder.get(), null);
 		return _this();
 	}