diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java index a141737ef2..6007d7b4c6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java +++ b/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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java index 2515f13333..72d657b7aa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java +++ b/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, 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, 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, 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, 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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java index 6d717da45b..f130177f26 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java +++ b/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, 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; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java index fd3f9ce979..e4bec90771 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java +++ b/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 { - private final Map propertyExpressions = new HashMap(); + private final Map propertyExpressions = new HashMap<>(); - private final Map> headerExpressions = - new HashMap>(); + private final Map> headerExpressions = new HashMap<>(); EnricherSpec() { super(new ContentEnricher()); @@ -167,15 +165,7 @@ public class EnricherSpec extends ConsumerEndpointSpec EnricherSpec property(String key, V value) { - this.propertyExpressions.put(key, new ValueExpression(value)); + this.propertyExpressions.put(key, new ValueExpression<>(value)); return _this(); } @@ -234,7 +224,7 @@ public class EnricherSpec extends ConsumerEndpointSpec EnricherSpec header(String name, V value) { - return this.header(name, value, null); + return header(name, value, null); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java index cc296f2a31..a6bca3b08b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java +++ b/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 @@ -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; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java index f879bbe960..d963b606df 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java +++ b/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(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 29b8043412..d1987e1f7c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -336,13 +336,27 @@ public abstract class IntegrationFlowDefinition 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 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 RecipientListRouterSpec recipientFlow(GenericSelector

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 */ public RouterSpec 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(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java index 945168a99a..c6639ebf6c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java +++ b/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 lifecycles = new LinkedList<>(); + private MessageChannel inputChannel; + private boolean running; StandardIntegrationFlow(Map 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 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowRegistration.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowRegistration.java index 8156052144..aaf016fb40 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowRegistration.java +++ b/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" + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index 9d8cf43fa9..6e60369fc8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/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) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java index 27d60926ac..9d4dddf42d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java +++ b/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 + .handle((p, h) -> p.toUpperCase()); + } + @Bean public IntegrationFlow routeSubflowToReplyChannelFlow() { return f -> f .route("true", m -> m - .subFlowMapping(true, sf -> sf - .handle((p, h) -> p.toUpperCase()) - ) + .subFlowMapping(true, upperCase()) ); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java index 0207dbc195..2d61f373f3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java +++ b/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 + .handle((p, h) -> someService().someServiceMethod(p)); + } + + @Bean + public IntegrationFlow replyProducingSubFlowEnricher() { return f -> f .enrich(e -> e.requestPayload(p -> p.getPayload().getName()) - .requestSubFlow(sf -> sf - .handle((p, h) -> someService.someServiceMethod(p))) + .requestSubFlow(someServiceFlow()) .headerFunction("foo", Message::getPayload) .propertyFunction("name", Message::getPayload)) .channel("subFlowTestReplyChannel");