Wrap non-StandardIntegrationFlow into Proxy (#2440)

* Wrap non-`StandardIntegrationFlow` into Proxy

In previous version all the `IntegrationFlow` beans have been replaced
by the `StandardIntegrationFlow` i the `IntegrationFlowBeanPostProcessor`
if they are lambda.
That works for Java, but doesn't with Kotlin, since lambdas i last one
are not synthetic classes.
Therefore some Java DSL definitions (especially `.subFlowMapping()`)
don't work consistently in two languages.

* Introduce `IntegrationFlowLifecycleAdvice` to wrap all the
non-`StandardIntegrationFlow`s (excluding `IntegrationFlowAdapter`)
into the `Proxy` to expose `SmartLifecycle` and `getInputChannel()`
operations and delegate them to the internal `StandardIntegrationFlow`
created by the `IntegrationFlowBeanPostProcessor`.
* This way any custom `IntegrationFlow` implementations can be used
for manual flow registration via `IntegrationFlowContext`
* Polish `RouterDslTests.kt` for the `@Bean`s for sub-flows.
* Document in the `dsl.adoc` a request-reply approach for the case
when `.subFlowMapping()` refers to an `IntegrationFlow` `@Bean`.
* Polishing for the `FlowServiceTests` since all the
non-`StandardIntegrationFlow`s and not-`IntegrationFlowAdapter`s are
wrapped now to the Proxy.
* Add missing `from()` delegations into the `IntegrationFlowAdapter`
* Polishing for the `ManualFlowTests` since all the `IntegrationFlow`
now are `Lifecycle` after wrapping to the Proxy.

* Add JavaDocs to the IntegrationFlowLifecycleAdvice and polishing for the dsl.adoc

* * Add JavaDocs for the `IntegrationFlowBeanPostProcessor.processIntegrationFlowImpl()`
* Improve JavaDoc for the `RouterSpec.subFlowMapping()`
* Assert in the `FlowServiceTests` that proxied custom flow implements
all the expected interfaces

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

@@ -18,11 +18,15 @@ package org.springframework.integration.dsl;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.reactivestreams.Publisher;
 
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
@@ -178,6 +182,27 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
 		return IntegrationFlows.from(service, methodName, endpointConfigurer);
 	}
 
+	protected <T> IntegrationFlowBuilder from(Supplier<T> messageSource) {
+		return IntegrationFlows.from(messageSource);
+	}
+
+	protected <T> IntegrationFlowBuilder from(Supplier<T> messageSource,
+			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+		return IntegrationFlows.from(messageSource, endpointConfigurer);
+	}
+
+	protected IntegrationFlowBuilder from(Class<?> serviceInterface) {
+		return IntegrationFlows.from(serviceInterface);
+	}
+
+	protected IntegrationFlowBuilder from(Class<?> serviceInterface, String beanName) {
+		return IntegrationFlows.from(serviceInterface, beanName);
+	}
+
+	protected IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {
+		return IntegrationFlows.from(publisher);
+	}
+
 	protected abstract IntegrationFlowDefinition<?> buildFlow();
 
 }

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

@@ -20,6 +20,8 @@ import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanCreationNotAllowedException;
 import org.springframework.beans.factory.BeanFactory;
@@ -33,6 +35,7 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.core.io.DescriptiveResource;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DirectChannel;
@@ -250,11 +253,44 @@ public class IntegrationFlowBeanPostProcessor
 		return flow;
 	}
 
+	/**
+	 * Only invoked for {@link IntegrationFlow} instances that are not
+	 * {@link StandardIntegrationFlow}s; typically lambdas. Creates a new
+	 * {@link StandardIntegrationFlow} with an input channel named {@code beanName.input}
+	 * and the flow defined by the flow parameter. If the flow is not an
+	 * {@link IntegrationFlowAdapter} the original, user-provided {@link IntegrationFlow}
+	 * is wrapped in a proxy and advised with a {@link IntegrationFlowLifecycleAdvice};
+	 * see its javadocs for more information.
+	 */
 	private Object processIntegrationFlowImpl(IntegrationFlow flow, String beanName) {
 		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(beanName + ".input");
+
 		flow.configure(flowBuilder);
-		Object standardIntegrationFlow = processStandardIntegrationFlow(flowBuilder.get(), beanName);
-		return isLambda(flow) ? standardIntegrationFlow : flow;
+
+		StandardIntegrationFlow target = flowBuilder.get();
+		processStandardIntegrationFlow(target, beanName);
+
+		if (!(flow instanceof IntegrationFlowAdapter)) {
+			NameMatchMethodPointcutAdvisor integrationFlowAdvice =
+					new NameMatchMethodPointcutAdvisor(new IntegrationFlowLifecycleAdvice(target));
+			integrationFlowAdvice.setMappedNames(
+					"getInputChannel",
+					"start",
+					"stop",
+					"isRunning",
+					"isAutoStartup",
+					"getPhase");
+
+			ProxyFactory proxyFactory = new ProxyFactory(flow);
+			proxyFactory.addAdvisor(integrationFlowAdvice);
+			if (!(flow instanceof SmartLifecycle)) {
+				proxyFactory.addInterface(SmartLifecycle.class);
+			}
+			return proxyFactory.getProxy(this.beanFactory.getBeanClassLoader());
+		}
+		else {
+			return flow;
+		}
 	}
 
 	private void processIntegrationComponentSpec(IntegrationComponentSpec<?, ?> bean) {
@@ -321,9 +357,4 @@ public class IntegrationFlowBeanPostProcessor
 		return id;
 	}
 
-	private static boolean isLambda(Object o) {
-		Class<?> aClass = o.getClass();
-		return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass();
-	}
-
 }

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

@@ -0,0 +1,115 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.context.SmartLifecycle;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * An AOP {@link MethodInterceptor} for the {@link IntegrationFlow} proxies
+ * with a delegation to an associated {@link StandardIntegrationFlow} instance, which
+ * is not exposed as a bean during a target {@link IntegrationFlow} bean processing.
+ *
+ * <p> In most cases an associated internal {@link StandardIntegrationFlow}
+ * exposes an {@code inputChannel} bean for the target {@link IntegrationFlow},
+ * which doesn't start from the channel, e.g. instantiated from lambda.
+ * This way the advice first tries to obtain an {@code inputChannel} from the
+ * target {@link IntegrationFlow} and then falls back to the {@link #delegate}.
+ *
+ * <p> Another aspect of this advice is to control and delegate {@link SmartLifecycle}
+ * of the target {@link IntegrationFlow} and associated {@link #delegate}.
+ * The {@link SmartLifecycle#start()} and {@link SmartLifecycle#stop()} operations
+ * are delegated to the {@link StandardIntegrationFlow} as is because that instance
+ * isn't controlled by the standard application context lifecycle.
+ * The {@link SmartLifecycle#isAutoStartup()}, {@link SmartLifecycle#getPhase()}
+ * and {@link SmartLifecycle#isRunning()} are called on the {@link #delegate}
+ * only in case when {@link MethodInvocation#proceed()} returns {@code null}
+ * or isn't called at all, e.g. when target {@link IntegrationFlow} doesn't
+ * implement {@link SmartLifecycle}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.1
+ */
+class IntegrationFlowLifecycleAdvice implements MethodInterceptor {
+
+	private final StandardIntegrationFlow delegate;
+
+	IntegrationFlowLifecycleAdvice(StandardIntegrationFlow delegate) {
+		this.delegate = delegate;
+	}
+
+	@Override
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		Object target = invocation.getThis();
+		String method = invocation.getMethod().getName();
+		Object result = null;
+
+		if ("getInputChannel".equals(method)) {
+			result = invocation.proceed();
+			if (result == null) {
+				result = this.delegate.getInputChannel();
+			}
+		}
+		else {
+			if (target instanceof SmartLifecycle) {
+				result = invocation.proceed();
+			}
+
+			switch (method) {
+
+				case "start":
+					this.delegate.start();
+					break;
+
+				case "stop":
+					Object[] arguments = invocation.getArguments();
+					if (!ObjectUtils.isEmpty(arguments)) {
+						this.delegate.stop((Runnable) arguments[0]);
+					}
+					else {
+						this.delegate.stop();
+					}
+					break;
+
+				case "isRunning":
+					if (result == null) {
+						result = this.delegate.isRunning();
+					}
+					break;
+
+				case "isAutoStartup":
+					if (result == null) {
+						result = this.delegate.isAutoStartup();
+					}
+					break;
+
+				case "getPhase":
+					if (result == null) {
+						result = this.delegate.getPhase();
+					}
+					break;
+			}
+		}
+
+		return result;
+	}
+
+}

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

@@ -143,6 +143,14 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
 	 * Add a subflow as an alternative to a {@link #channelMapping(Object, String)}.
 	 * {@link #prefix(String)} and {@link #suffix(String)} cannot be used when subflow
 	 * mappings are used.
+	 * <p> If subflow should refer to the external {@link IntegrationFlow} bean and
+	 * there is a requirement to expect reply from there, such a reference should be
+	 * wrapped with a {@code .gateway()}:
+	 * <pre class="code">
+	 * {@code
+	 *     .subFlowMapping(false, sf -> sf.gateway(evenFlow())))
+	 * }
+	 * </pre>
 	 * @param key the key.
 	 * @param subFlow the subFlow.
 	 * @return the router spec.

spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java

@@ -16,8 +16,11 @@
 
 package org.springframework.integration.dsl.flowservices;
 
+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.Collection;
 import java.util.Collections;
@@ -30,11 +33,15 @@ import java.util.stream.Collectors;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.aop.framework.Advised;
+import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.Ordered;
 import org.springframework.integration.annotation.Aggregator;
 import org.springframework.integration.annotation.CorrelationStrategy;
 import org.springframework.integration.annotation.Filter;
@@ -70,21 +77,29 @@ import org.springframework.util.StringUtils;
 @DirtiesContext
 public class FlowServiceTests {
 
+	@Autowired(required = false)
+	@Qualifier("flowServiceTests.MyFlow")
+	private IntegrationFlow myFlow;
+
 	@Autowired
 	@Qualifier("flowServiceTests.MyFlow.input")
 	private MessageChannel input;
 
-	@Autowired(required = false)
-	private MyFlow myFlow;
-
 	@Autowired
 	private PollableChannel myFlowAdapterOutput;
 
 	@Test
-	public void testFlowServiceAndLogAsLastNoError() {
+	public void testFlowServiceAndLogAsLastNoError() throws Exception {
 		assertNotNull(this.myFlow);
+		assertTrue(AopUtils.isAopProxy(this.myFlow));
+		assertThat(this.myFlow, instanceOf(Advised.class));
+		assertThat(this.myFlow, instanceOf(Ordered.class));
+		assertThat(this.myFlow, instanceOf(SmartLifecycle.class));
+
 		this.input.send(MessageBuilder.withPayload("foo").build());
-		Object result = this.myFlow.resultOverLoggingHandler.get();
+
+		MyFlow myFlow = (MyFlow) ((Advised) this.myFlow).getTargetSource().getTarget();
+		Object result = myFlow.resultOverLoggingHandler.get();
 		assertNotNull(result);
 		assertEquals("FOO", result);
 	}
@@ -134,7 +149,7 @@ public class FlowServiceTests {
 	}
 
 	@Component
-	public static class MyFlow implements IntegrationFlow {
+	public static class MyFlow implements IntegrationFlow, Ordered {
 
 		private final AtomicReference<Object> resultOverLoggingHandler = new AtomicReference<>();
 
@@ -147,6 +162,11 @@ public class FlowServiceTests {
 					});
 		}
 
+		@Override
+		public int getOrder() {
+			return 0;
+		}
+
 	}
 
 	@Component

spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java

@@ -204,32 +204,6 @@ public class ManualFlowTests {
 
 	@Test
 	public void testWrongLifecycle() {
-
-		class MyIntegrationFlow implements IntegrationFlow {
-
-			@Override
-			public void configure(IntegrationFlowDefinition<?> flow) {
-				flow.bridge();
-			}
-
-		}
-
-		IntegrationFlow testFlow = new MyIntegrationFlow();
-
-		// This is fine because we are not going to start it automatically.
-		assertNotNull(this.integrationFlowContext.registration(testFlow)
-				.autoStartup(false)
-				.register());
-
-		try {
-			this.integrationFlowContext.registration(testFlow).register();
-			fail("IllegalStateException expected");
-		}
-		catch (Exception e) {
-			assertThat(e, instanceOf(IllegalStateException.class));
-			assertThat(e.getMessage(), containsString("Consider to implement it for [" + testFlow + "]."));
-		}
-
 		try {
 			this.integrationFlowContext.remove("foo");
 			fail("IllegalStateException expected");

spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt

@@ -25,14 +25,17 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.integration.channel.QueueChannel
 import org.springframework.integration.config.EnableIntegration
 import org.springframework.integration.dsl.IntegrationFlow
+import org.springframework.integration.support.MessageBuilder
 import org.springframework.messaging.MessageChannel
 import org.springframework.messaging.PollableChannel
 import org.springframework.messaging.support.GenericMessage
 import org.springframework.test.annotation.DirtiesContext
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
 
+
 /**
  * @author Artem Bilan
  *
@@ -66,6 +69,30 @@ class RouterDslTests {
     }
 
 
+    @Autowired
+    @Qualifier("splitRouteAggregate.input")
+    private lateinit var splitRouteAggregateInput: MessageChannel
+
+    @Test
+    fun `route to two subflows using them as bean references`() {
+
+        val replyChannel = QueueChannel()
+        val message = MessageBuilder.withPayload(arrayOf(1, 2, 3))
+                .setReplyChannel(replyChannel)
+                .build()
+
+        this.splitRouteAggregateInput.send(message)
+
+        val receive = replyChannel.receive(10000)
+
+        val payload = receive?.payload
+
+        assert(payload).isNotNull {
+            it.isInstanceOf(List::class.java)
+            it.isEqualTo(listOf("even", "odd", "even"))
+        }
+    }
+
     @Configuration
     @EnableIntegration
     class Config {
@@ -83,6 +110,29 @@ class RouterDslTests {
                             .channel { c -> c.queue("routerTwoSubFlowsOutput") }
                 }
 
+        @Bean
+        fun splitRouteAggregate() =
+                IntegrationFlow { f ->
+                    f.split()
+                            .route<Int, Boolean>({ o -> o % 2 == 0 },
+                                    { m ->
+                                        m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
+                                                .subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
+                                    })
+                            .aggregate()
+                }
+
+        @Bean
+        fun oddFlow() =
+                IntegrationFlow { flow ->
+                    flow.handle<Any> { _, _ -> "odd" }
+                }
+
+        @Bean
+        fun evenFlow() =
+                IntegrationFlow { flow ->
+                    flow.handle<Any> { _, _ -> "even" }
+                }
     }
 
 }

src/reference/asciidoc/dsl.adoc

@@ -578,6 +578,8 @@ The result of this definition is the same bunch of Integration components wired
 Only limitation is here, that this flow is started with named direct channel - `lambdaFlow.input`.
 And Lambda flow can't start from `MessageSource` or `MessageProducer`.
 
+Starting _version 5.1_, this kind of `IntegrationFlow` are wrapped to the proxy for exposing lifecycle control and provide access to the `inputChannel` of the internally associated `StandardIntegrationFlow`.
+
 [[java-dsl-function-expression]]
 === FunctionExpression
 
@@ -643,6 +645,42 @@ public IntegrationFlow routeFlow() {
 The `.channelMapping()` continues to work as in regular `Router` mapping, but the `.subFlowMapping()` tied that subflow with main flow.
 In other words, any router's subflow returns to the main flow after `.route()`.
 
+[IMPORTANT]
+=====
+Sometimes it is necessary to refer to an existing `IntegrationFlow` `@Bean` from the `.subFlowMapping()`:
+[source,java]
+----
+@Bean
+public IntegrationFlow splitRouteAggregate() {
+    return f -> f
+            .split()
+            .<Integer, Boolean>route(o -> o % 2 == 0,
+                    m -> m
+                            .subFlowMapping(true, oddFlow())
+                            .subFlowMapping(false, sf -> sf.gateway(evenFlow())))
+            .aggregate();
+}
+
+@Bean
+public IntegrationFlow oddFlow() {
+    return f -> f.handle(m -> System.out.println("odd"));
+}
+
+@Bean
+public IntegrationFlow evenFlow() {
+    return f -> f.handle((p, h) -> "even");
+}
+----
+
+In this case, when you need to receive a reply from such a sub-flow and continue the main flow, this `IntegrationFlow` bean reference (or its input channel) has to be wrapped with a `.gateway()` as shown above.
+The `oddFlow()` referece in the sample above is not wrapped to the `.gateway()` - therefore we don't expect a reply from this routing branch.
+Otherwise you end up with an exception like:
+....
+Caused by: org.springframework.beans.factory.BeanCreationException: The 'currentComponent' (org.springframework.integration.router.MethodInvokingRouter@7965a51c) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow.
+....
+When a sub-flow is configured as a lambda, the Framework handles the request-reply interaction with the sub-flow and a gateway is not needed.
+=====
+
 Of course, subflows can be nested with any depth, but we don't recommend to do that because, in fact, even in the router case, adding complex subflows within a flow would quickly begin to look like a plate of spaghetti and difficult for a human to parse.
 
 [[java-dsl-protocol-adapters]]