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()`

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

@@ -82,11 +82,6 @@ public class IntegrationFlowBeanPostProcessor
 		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/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 class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 	 * 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 class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 	 * 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,17 +84,9 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 	 * 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));
 	}
 
 	/**
@@ -106,7 +94,6 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 	 * 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 java.util.function.Function;
 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 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
 		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.Map;
 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 @@ import reactor.util.function.Tuple2;
  */
 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 class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
 	 * @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 class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
 	 * @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 class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
 	 * @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 final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpoin
 	 * @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

@@ -69,6 +69,12 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
 		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/IntegrationFlowDefinition.java

@@ -336,13 +336,27 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @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
@@ -2166,11 +2180,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @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);
 	}
 
@@ -2702,9 +2713,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		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

@@ -21,6 +21,8 @@ import java.util.Map;
 import java.util.concurrent.Executor;
 
 import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
 
 /**
  * @author Artem Bilan
@@ -44,11 +46,21 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSub
 		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();
 	}

spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.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.
@@ -17,7 +17,6 @@
 package org.springframework.integration.dsl;
 
 import org.springframework.expression.Expression;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
@@ -25,7 +24,6 @@ import org.springframework.integration.filter.MethodInvokingSelector;
 import org.springframework.integration.handler.LambdaMessageProcessor;
 import org.springframework.integration.router.RecipientListRouter;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
@@ -204,8 +202,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
 	 * @return the router spec.
 	 */
 	public <P> RecipientListRouterSpec recipientFlow(GenericSelector<P> selector, IntegrationFlow subFlow) {
-		Assert.notNull(subFlow, "'subFlow' must not be null");
-		DirectChannel channel = populateSubFlow(subFlow);
+		MessageChannel channel = obtainInputChannelFromFlow(subFlow);
 		return recipient(channel, selector);
 	}
 
@@ -213,7 +210,6 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
 	 * Adds a subflow that will be invoked as a recipient.
 	 * @param subFlow the subflow.
 	 * @return the router spec.
-	 * @since 1.2
 	 */
 	public RecipientListRouterSpec recipientFlow(IntegrationFlow subFlow) {
 		return recipientFlow((String) null, subFlow);
@@ -235,20 +231,10 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
 	 * @param expression the expression.
 	 * @param subFlow the subflow.
 	 * @return the router spec.
-	 * @since 1.2
 	 */
 	public RecipientListRouterSpec recipientFlow(Expression expression, IntegrationFlow subFlow) {
-		Assert.notNull(subFlow, "'subFlow' must not be null");
-		DirectChannel channel = populateSubFlow(subFlow);
+		MessageChannel channel = obtainInputChannelFromFlow(subFlow);
 		return recipient(channel, expression);
 	}
 
-	private DirectChannel populateSubFlow(IntegrationFlow subFlow) {
-		DirectChannel channel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
-		subFlow.configure(flowBuilder);
-		this.componentsToRegister.put(flowBuilder.get(), null);
-		return channel;
-	}
-
 }

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

@@ -21,11 +21,11 @@ import java.util.Map;
 
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.router.AbstractMappingMessageRouter;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.integration.support.management.MappingMessageRouterManagement;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -149,17 +149,16 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
 	 */
 	public RouterSpec<K, R> subFlowMapping(K key, IntegrationFlow subFlow) {
 		Assert.notNull(key, "'key' must not be null");
-		Assert.notNull(subFlow, "'subFlow' must not be null");
 		Assert.state(!(StringUtils.hasText(this.prefix) || StringUtils.hasText(this.suffix)),
 				"The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive");
 
-		DirectChannel channel = new DirectChannel();
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
-		subFlow.configure(flowBuilder);
+		MessageChannel channel = obtainInputChannelFromFlow(subFlow, false);
 
-		this.componentsToRegister.put(flowBuilder, null);
+		Assert.isInstanceOf(NamedComponent.class, channel,
+				() -> "The routing channel '" + channel +
+						"' from the flow '" + subFlow + "' must be instance of 'NamedComponent'.");
 
-		this.mappingProvider.addMapping(key, channel);
+		this.mappingProvider.addMapping(key, (NamedComponent) channel);
 		return _this();
 	}
 

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

@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.context.SmartLifecycle;
+import org.springframework.messaging.MessageChannel;
 
 /**
  * The standard implementation of the {@link IntegrationFlow} interface instantiated
@@ -69,12 +70,35 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
 
 	private final List<SmartLifecycle> lifecycles = new LinkedList<>();
 
+	private MessageChannel inputChannel;
+
 	private boolean running;
 
 	StandardIntegrationFlow(Map<Object, String> integrationComponents) {
 		this.integrationComponents = new LinkedHashMap<>(integrationComponents);
 	}
 
+	@Override
+	public void configure(IntegrationFlowDefinition<?> flow) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public MessageChannel getInputChannel() {
+		if (this.inputChannel == null) {
+			this.inputChannel =
+					this.integrationComponents.keySet()
+							.stream()
+							.filter(MessageChannel.class::isInstance)
+							.map(MessageChannel.class::cast)
+							.findFirst()
+							.orElseThrow(() -> new IllegalStateException("The 'IntegrationFlow' [" + this + "] " +
+									"doesn't start with 'MessageChannel' for direct message sending."));
+		}
+
+		return this.inputChannel;
+	}
+
 	public void setIntegrationComponents(Map<Object, String> integrationComponents) {
 		this.integrationComponents.clear();
 		this.integrationComponents.putAll(integrationComponents);
@@ -84,11 +108,6 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
 		return Collections.unmodifiableMap(this.integrationComponents);
 	}
 
-	@Override
-	public void configure(IntegrationFlowDefinition<?> flow) {
-		throw new UnsupportedOperationException();
-	}
-
 	@Override
 	public void start() {
 		if (!this.running) {

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

@@ -20,7 +20,6 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.StandardIntegrationFlow;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 
@@ -79,19 +78,9 @@ public class IntegrationFlowRegistration {
 
 	public MessageChannel getInputChannel() {
 		if (this.inputChannel == null) {
-			if (this.integrationFlow instanceof StandardIntegrationFlow) {
-				StandardIntegrationFlow integrationFlow = (StandardIntegrationFlow) this.integrationFlow;
-				Object next = integrationFlow.getIntegrationComponents().keySet().iterator().next();
-				if (next instanceof MessageChannel) {
-					this.inputChannel = (MessageChannel) next;
-				}
-				else {
-					throw new IllegalStateException("The 'IntegrationFlow' [" + integrationFlow + "] " +
-							"doesn't start with 'MessageChannel' for direct message sending.");
-				}
-			}
-			else {
-				throw new IllegalStateException("Only 'StandardIntegrationFlow' instances " +
+			this.inputChannel = this.integrationFlow.getInputChannel();
+			if (this.inputChannel == null) {
+				throw new IllegalStateException("Only 'IntegrationFlow' instances started from the 'MessageChannel' " +
 						"(e.g. extracted from 'IntegrationFlow' Lambdas) can be used " +
 						"for direct 'send' operation. " +
 						"But [" + this.integrationFlow + "] ins't one of them.\n" +

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

@@ -266,7 +266,7 @@ public class IntegrationFlowTests {
 		catch (Exception e) {
 			assertThat(e, instanceOf(BeanCreationException.class));
 			assertThat(e.getMessage(), containsString("'.fixedSubscriberChannel()' " +
-					"can't be the last EIP-method in the IntegrationFlow definition"));
+					"can't be the last EIP-method in the 'IntegrationFlow' definition"));
 		}
 		finally {
 			if (context != null) {

spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java

@@ -553,13 +553,17 @@ public class RouterTests {
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow upperCase() {
+			return f -> f
+					.<String>handle((p, h) -> p.toUpperCase());
+		}
+
 		@Bean
 		public IntegrationFlow routeSubflowToReplyChannelFlow() {
 			return f -> f
 					.<Boolean>route("true", m -> m
-							.subFlowMapping(true, sf -> sf
-									.<String>handle((p, h) -> p.toUpperCase())
-							)
+							.subFlowMapping(true, upperCase())
 					);
 		}
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java

@@ -376,11 +376,16 @@ public class TransformerTests {
 		}
 
 		@Bean
-		public IntegrationFlow replyProducingSubFlowEnricher(SomeService someService) {
+		public IntegrationFlow someServiceFlow() {
+			return f -> f
+					.<String>handle((p, h) -> someService().someServiceMethod(p));
+		}
+
+		@Bean
+		public IntegrationFlow replyProducingSubFlowEnricher() {
 			return f -> f
 					.enrich(e -> e.<TestPojo>requestPayload(p -> p.getPayload().getName())
-							.requestSubFlow(sf -> sf
-									.<String>handle((p, h) -> someService.someServiceMethod(p)))
+							.requestSubFlow(someServiceFlow())
 							.<String>headerFunction("foo", Message::getPayload)
 							.propertyFunction("name", Message::getPayload))
 					.channel("subFlowTestReplyChannel");