diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java index 9d712daa09..efcaf8f700 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java @@ -40,7 +40,6 @@ import com.rabbitmq.client.Channel; import com.rabbitmq.client.Command; import com.rabbitmq.client.ConfirmListener; import com.rabbitmq.client.Consumer; -import com.rabbitmq.client.FlowListener; import com.rabbitmq.client.GetResponse; import com.rabbitmq.client.Method; import com.rabbitmq.client.ReturnListener; @@ -181,13 +180,13 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { public void addReturnListener(ReturnListener listener) { } - @SuppressWarnings("unused") - public FlowListener getFlowListener() { + @SuppressWarnings("deprecation") + public com.rabbitmq.client.FlowListener getFlowListener() { return null; } - @SuppressWarnings("unused") - public void setFlowListener(FlowListener listener) { + @SuppressWarnings("deprecation") + public void setFlowListener(com.rabbitmq.client.FlowListener listener) { } @SuppressWarnings("unused") @@ -497,12 +496,12 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { @Override @Deprecated - public void addFlowListener(FlowListener listener) { + public void addFlowListener(com.rabbitmq.client.FlowListener listener) { } @Override @Deprecated - public boolean removeFlowListener(FlowListener listener) { + public boolean removeFlowListener(com.rabbitmq.client.FlowListener listener) { return false; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 3941e99b8e..f1919d181f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -93,7 +93,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private final Map> expireGroupScheduledFutures = new HashMap<>(); - private final MessageGroupProcessor outputProcessor; + private MessageGroupProcessor outputProcessor; private volatile MessageGroupStore messageStore; @@ -133,9 +133,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { - Assert.notNull(processor); + Assert.notNull(processor, "'processor' must not be null"); + Assert.notNull(store, "'store' must not be null"); - Assert.notNull(store); setMessageStore(store); this.outputProcessor = processor; this.correlationStrategy = (correlationStrategy == null @@ -187,8 +187,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.forceReleaseAdviceChain = forceReleaseAdviceChain; } - public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { - this.evaluationContext = evaluationContext; + /** + * Specify a {@link MessageGroupProcessor} for the output function. + * @param outputProcessor the {@link MessageGroupProcessor} to use + * @since 5.0 + */ + public void setOutputProcessor(MessageGroupProcessor outputProcessor) { + Assert.notNull(outputProcessor, "'processor' must not be null"); + this.outputProcessor = outputProcessor; } @Override 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 72e49e42c0..58f505f365 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 @@ -38,7 +38,7 @@ import org.springframework.util.Assert; public class AbstractRouterSpec, R extends AbstractMessageRouter> extends MessageHandlerSpec implements ComponentsRegistration { - protected final List subFlows = new ArrayList(); + protected final List componentsToRegister = new ArrayList<>(); private boolean defaultToParentFlow; @@ -102,7 +102,7 @@ public class AbstractRouterSpec, R extends Ab IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel); subFlow.configure(flowBuilder); - this.subFlows.add(flowBuilder); + this.componentsToRegister.add(flowBuilder); return defaultOutputChannel(channel); } @@ -125,7 +125,7 @@ public class AbstractRouterSpec, R extends Ab @Override public Collection getComponentsToRegister() { - return this.subFlows; + return this.componentsToRegister; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java index 5364ad21f8..97ccee418a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java @@ -16,16 +16,11 @@ package org.springframework.integration.dsl; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor; import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; -import org.springframework.integration.store.MessageGroup; -import org.springframework.util.Assert; /** * A {@link CorrelationHandlerSpec} for an {@link AggregatingMessageHandler}. @@ -37,7 +32,7 @@ import org.springframework.util.Assert; public class AggregatorSpec extends CorrelationHandlerSpec { AggregatorSpec() { - super(new InternalAggregatingMessageHandler()); + super(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor())); } /** @@ -45,7 +40,7 @@ public class AggregatorSpec extends CorrelationHandlerSpec, H extends AbstractCorrelatingMessageHandler> +public abstract class CorrelationHandlerSpec, + H extends AbstractCorrelatingMessageHandler> extends ConsumerEndpointSpec { private final List forceReleaseAdviceChain = new LinkedList(); @@ -166,7 +166,7 @@ public abstract class * Configure the handler with {@link org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy} * and {@link org.springframework.integration.aggregator.MethodInvokingReleaseStrategy} using the target * object which should have methods annotated appropriately for each function. - * @param target the target object, + * @param target the target object * @return the handler spec. * @see AbstractCorrelatingMessageHandler#setCorrelationStrategy(CorrelationStrategy) * @see AbstractCorrelatingMessageHandler#setReleaseStrategy(ReleaseStrategy) @@ -188,8 +188,8 @@ public abstract class } /** - * Configure the handler with an {@link ExpressionEvaluatingCorrelationStrategy} for the - * given expression. + * Configure the handler with an {@link ExpressionEvaluatingCorrelationStrategy} + * for the given expression. * @param correlationExpression the correlation expression. * @return the handler spec. * @see AbstractCorrelatingMessageHandler#setCorrelationStrategy(CorrelationStrategy) @@ -233,6 +233,7 @@ public abstract class /** * Configure the handler with an {@link ExpressionEvaluatingReleaseStrategy} for the * given expression. + * * @param releaseExpression the correlation expression. * @return the handler spec. * @see AbstractCorrelatingMessageHandler#setReleaseStrategy(ReleaseStrategy) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/DslRecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/DslRecipientListRouter.java deleted file mode 100644 index a174409231..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/DslRecipientListRouter.java +++ /dev/null @@ -1,138 +0,0 @@ -/* - * Copyright 2016 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 java.util.ArrayList; -import java.util.List; - -import org.springframework.beans.BeansException; -import org.springframework.expression.Expression; -import org.springframework.integration.core.GenericSelector; -import org.springframework.integration.core.MessageSelector; -import org.springframework.integration.dsl.support.MessageChannelReference; -import org.springframework.integration.filter.ExpressionEvaluatingSelector; -import org.springframework.integration.filter.MethodInvokingSelector; -import org.springframework.integration.router.RecipientListRouter; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.core.DestinationResolutionException; -import org.springframework.util.StringUtils; - -import reactor.util.function.Tuple2; -import reactor.util.function.Tuples; - -/** - * @author Artem Bilan - * - * TODO Move the logic to the RecipientListRouter - */ -class DslRecipientListRouter extends RecipientListRouter { - - private final List> recipients = new ArrayList<>(); - - void add(String channelName, Expression expression) { - this.recipients.add(Tuples.of(channelName, expression)); - } - - void add(String channelName, GenericSelector selector) { - this.recipients.add(Tuples.of(channelName, selector)); - } - - void add(MessageChannel channel, Expression expression) { - this.recipients.add(Tuples.of(channel, expression)); - } - - void add(MessageChannel channel, GenericSelector selector) { - this.recipients.add(Tuples.of(channel, selector)); - } - - @Override - public void onInit() throws Exception { - List recipients = new ArrayList(this.recipients.size()); - - for (Tuple2 recipient : this.recipients) { - if (recipient.getT1() instanceof String) { - recipients.add(new DslRecipient(new MessageChannelReference((String) recipient.getT1()), - populateRecipientSelector(recipient.getT2()))); - } - else { - recipients.add(new Recipient((MessageChannel) recipient.getT1(), - populateRecipientSelector(recipient.getT2()))); - } - } - - setRecipients(recipients); - this.recipients.clear(); - super.onInit(); - } - - private MessageSelector populateRecipientSelector(final Object recipientSelector) { - if (recipientSelector instanceof String) { - String expression = (String) recipientSelector; - if (StringUtils.hasText(expression)) { - ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression); - selector.setBeanFactory(getBeanFactory()); - return selector; - } - } - else if (recipientSelector instanceof Expression) { - ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector((Expression) recipientSelector); - selector.setBeanFactory(getBeanFactory()); - return selector; - } - else if (recipientSelector instanceof MessageSelector) { - return (MessageSelector) recipientSelector; - } - else if (recipientSelector instanceof GenericSelector) { - return new MethodInvokingSelector(new LambdaMessageProcessor(recipientSelector, null)); - } - return null; - } - - private class DslRecipient extends Recipient { - - private volatile MessageChannel channel; - - DslRecipient(MessageChannelReference channel, MessageSelector selector) { - super(channel, selector); - } - - @Override - public MessageChannel getChannel() { - if (this.channel == null) { - synchronized (this) { - if (this.channel == null) { - this.channel = resolveChannelName((MessageChannelReference) super.getChannel()); - } - } - } - return this.channel; - } - - private MessageChannel resolveChannelName(MessageChannelReference channelReference) { - String channelName = channelReference.getName(); - try { - return DslRecipientListRouter.this.getBeanFactory().getBean(channelName, MessageChannel.class); - } - catch (BeansException e) { - throw new DestinationResolutionException("Failed to look up MessageChannel with name '" - + channelName + "' in the BeanFactory."); - } - } - - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java index 3a90e73954..5ce163bfd7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java @@ -16,6 +16,7 @@ package org.springframework.integration.dsl; +import org.springframework.integration.gateway.GatewayMessageHandler; import org.springframework.messaging.MessageChannel; /** 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 a58eeaabe2..1d49312317 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 @@ -54,12 +54,14 @@ import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; import org.springframework.integration.filter.MethodInvokingSelector; +import org.springframework.integration.gateway.GatewayMessageHandler; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.BeanNameMessageProcessor; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.handler.DelayHandler; import org.springframework.integration.handler.ExpressionCommandMessageProcessor; import org.springframework.integration.handler.GenericHandler; +import org.springframework.integration.handler.LambdaMessageProcessor; import org.springframework.integration.handler.LoggingHandler; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.handler.MessageTriggerAction; @@ -1617,7 +1619,7 @@ public abstract class IntegrationFlowDefinition B split(S splitter, Consumer> endpointConfigurer) { Assert.notNull(splitter); - return this.register(new SplitterEndpointSpec(splitter), endpointConfigurer); + return this.register(new SplitterEndpointSpec<>(splitter), endpointConfigurer); } /** @@ -1730,31 +1732,6 @@ public abstract class IntegrationFlowDefinition - * {@code - * .resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"), - * e -> e.phase(100)) - * } - * - * @param resequencerConfigurer the {@link Consumer} to provide {@link ResequencingMessageHandler} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - * @see GenericEndpointSpec - * @deprecated since 1.1 in favor of {@link #resequence(Consumer)} - */ - @Deprecated - public B resequence(Consumer resequencerConfigurer, - Consumer> endpointConfigurer) { - Assert.notNull(resequencerConfigurer); - ResequencerSpec spec = new ResequencerSpec(); - resequencerConfigurer.accept(spec); - return handle(spec.get().getT2(), endpointConfigurer); - } - /** * Populate the {@link ResequencingMessageHandler} with provided options from {@link ResequencerSpec}. * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. @@ -1782,31 +1759,6 @@ public abstract class IntegrationFlowDefinition - * {@code - * .aggregate(a -> a.correlationExpression("1").releaseStrategy(g -> g.size() == 25), - * e -> e.applySequence(false)) - * } - * - * @param aggregatorConfigurer the {@link Consumer} to provide {@link AggregatingMessageHandler} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - * @see GenericEndpointSpec - * @deprecated since 1.1 in favor of {@link #aggregate(Consumer)} - */ - @Deprecated - public B aggregate(Consumer aggregatorConfigurer, - Consumer> endpointConfigurer) { - Assert.notNull(aggregatorConfigurer); - AggregatorSpec spec = new AggregatorSpec(); - aggregatorConfigurer.accept(spec); - return this.handle(spec.get().getT2(), endpointConfigurer); - } - /** * Populate the {@link AggregatingMessageHandler} with provided options from {@link AggregatorSpec}. * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. @@ -1845,7 +1797,8 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { + public B route(String beanName, String method, + Consumer> routerConfigurer) { return route(beanName, method, routerConfigurer, null); } @@ -2184,7 +2137,7 @@ public abstract class IntegrationFlowDefinition B route(R router, Consumer> routerConfigurer, Consumer> endpointConfigurer) { - RouterSpec routerSpec = new RouterSpec(router); + RouterSpec routerSpec = new RouterSpec<>(router); if (routerConfigurer != null) { routerConfigurer.accept(routerSpec); } @@ -2466,7 +2419,7 @@ public abstract class IntegrationFlowDefinition B log(Function, Object> function) { Assert.notNull(function); - return log(new FunctionExpression>(function)); + return log(new FunctionExpression<>(function)); } /** @@ -2560,7 +2513,7 @@ public abstract class IntegrationFlowDefinition B log(LoggingHandler.Level level, String category, Function, Object> function) { Assert.notNull(function); - return log(level, category, new FunctionExpression>(function)); + return log(level, category, new FunctionExpression<>(function)); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java index fa4b54402a..c217e4b239 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java @@ -20,8 +20,10 @@ 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; +import org.springframework.integration.filter.MethodInvokingSelector; +import org.springframework.integration.handler.LambdaMessageProcessor; import org.springframework.integration.router.RecipientListRouter; -import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -36,7 +38,7 @@ import org.springframework.util.StringUtils; public class RecipientListRouterSpec extends AbstractRouterSpec { RecipientListRouterSpec() { - super(new DslRecipientListRouter()); + super(new RecipientListRouter()); } /** @@ -55,7 +57,13 @@ public class RecipientListRouterSpec extends AbstractRouterSpec>) selector); + return recipient(channelName, selector); } /** @@ -89,11 +97,24 @@ public class RecipientListRouterSpec extends AbstractRouterSpec RecipientListRouterSpec recipient(String channelName, GenericSelector

selector) { - Assert.hasText(channelName); - ((DslRecipientListRouter) this.target).add(channelName, selector); + MessageSelector messageSelector; + if (selector instanceof MessageSelector) { + messageSelector = (MessageSelector) selector; + } + else { + messageSelector = isLambda(selector) + ? new MethodInvokingSelector(new LambdaMessageProcessor(selector, null)) + : new MethodInvokingSelector(selector); + } + this.target.addRecipient(channelName, messageSelector); return _this(); } + private static boolean isLambda(Object o) { + Class aClass = o.getClass(); + return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass(); + } + /** * Adds a recipient channel that always will be selected. * @param channel the recipient channel. @@ -120,8 +141,9 @@ public class RecipientListRouterSpec extends AbstractRouterSpec>) selector); + return recipient(channel, selector); } /** @@ -143,8 +165,14 @@ public class RecipientListRouterSpec extends AbstractRouterSpec RecipientListRouterSpec recipient(MessageChannel channel, GenericSelector

selector) { - Assert.notNull(channel); - ((DslRecipientListRouter) this.target).add(channel, selector); + MessageSelector messageSelector; + if (selector instanceof MessageSelector) { + messageSelector = (MessageSelector) selector; + } + else { + messageSelector = new MethodInvokingSelector(new LambdaMessageProcessor(selector, null)); + } + this.target.addRecipient(channel, messageSelector); return _this(); } @@ -155,7 +183,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec>) selector, subFlow); + return recipientFlow(selector, subFlow); } /** @@ -168,8 +196,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec RecipientListRouterSpec recipientFlow(GenericSelector

selector, IntegrationFlow subFlow) { Assert.notNull(subFlow); DirectChannel channel = populateSubFlow(subFlow); - ((DslRecipientListRouter) this.target).add(channel, selector); - return _this(); + return recipient(channel, selector); } /** @@ -203,15 +230,14 @@ public class RecipientListRouterSpec extends AbstractRouterSpec * @see AbstractMappingMessageRouter#setPrefix(String) */ public RouterSpec prefix(String prefix) { - Assert.state(this.subFlows.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + Assert.state(this.componentsToRegister.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); this.prefix = prefix; this.target.setPrefix(prefix); return _this(); @@ -86,7 +86,7 @@ public final class RouterSpec * @see AbstractMappingMessageRouter#setSuffix(String) */ public RouterSpec suffix(String suffix) { - Assert.state(this.subFlows.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + Assert.state(this.componentsToRegister.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); this.suffix = suffix; this.target.setSuffix(suffix); return _this(); @@ -140,7 +140,7 @@ public final class RouterSpec IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel); subFlow.configure(flowBuilder); - this.subFlows.add(flowBuilder); + this.componentsToRegister.add(flowBuilder); this.mappingProvider.addMapping(key, channel); return _this(); @@ -150,7 +150,7 @@ public final class RouterSpec public Collection getComponentsToRegister() { // The 'mappingProvider' must be added to the 'componentToRegister' in the end to // let all other components to be registered before the 'RouterMappingProvider.onInit()' logic. - this.subFlows.add(this.mappingProvider); + this.componentsToRegister.add(this.mappingProvider); return super.getComponentsToRegister(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java similarity index 89% rename from spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java rename to spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java index 78e452273e..f054f815b8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java @@ -14,24 +14,24 @@ * limitations under the License. */ -package org.springframework.integration.dsl; +package org.springframework.integration.gateway; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.Lifecycle; -import org.springframework.integration.gateway.GatewayProxyFactoryBean; -import org.springframework.integration.gateway.RequestReplyExchanger; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; /** + * The {@link AbstractReplyProducingMessageHandler} implementation for mid-flow Gateway. + * * @author Artem Bilan * * @since 5.0 */ -class GatewayMessageHandler extends AbstractReplyProducingMessageHandler implements Lifecycle { +public class GatewayMessageHandler extends AbstractReplyProducingMessageHandler implements Lifecycle { private final GatewayProxyFactoryBean gatewayProxyFactoryBean; @@ -39,16 +39,16 @@ class GatewayMessageHandler extends AbstractReplyProducingMessageHandler impleme private volatile boolean running; - GatewayMessageHandler() { + public GatewayMessageHandler() { this.gatewayProxyFactoryBean = new GatewayProxyFactoryBean(); this.gatewayProxyFactoryBean.setServiceInterface(RequestReplyExchanger.class); } - void setRequestChannel(MessageChannel requestChannel) { + public void setRequestChannel(MessageChannel requestChannel) { this.gatewayProxyFactoryBean.setDefaultRequestChannel(requestChannel); } - void setRequestChannelName(String requestChannel) { + public void setRequestChannelName(String requestChannel) { this.gatewayProxyFactoryBean.setDefaultRequestChannelName(requestChannel); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java similarity index 90% rename from spring-integration-core/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java index 4330493aa0..ab61377780 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl; +package org.springframework.integration.handler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; @@ -28,7 +28,6 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.TypeDescriptor; import org.springframework.core.convert.support.DefaultConversionService; -import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -36,11 +35,14 @@ import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; /** + * The {@link MessageProcessor} implementation for method invocation on the single method classes + * - functional interface implementations. + * * @author Artem Bilan * * @since 5.0 */ -class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAware { +public class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAware { private final Object target; @@ -50,10 +52,9 @@ class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAwa private final Class[] parameterTypes; - private ConversionService conversionService; - LambdaMessageProcessor(Object target, Class payloadType) { + public LambdaMessageProcessor(Object target, Class payloadType) { Assert.notNull(target); this.target = target; final AtomicReference methodValue = new AtomicReference<>(); @@ -79,7 +80,7 @@ class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAwa public void setBeanFactory(BeanFactory beanFactory) throws BeansException { ConversionService conversionService = IntegrationUtils.getConversionService(beanFactory); if (conversionService == null) { - conversionService = new DefaultConversionService(); + conversionService = DefaultConversionService.getSharedInstance(); } this.conversionService = conversionService; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java index 8c079b5b21..b9136fb3dc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java @@ -16,7 +16,6 @@ package org.springframework.integration.router; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Iterator; @@ -24,8 +23,10 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Properties; +import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.stream.Collectors; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.core.MessageSelector; @@ -34,6 +35,7 @@ import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.core.DestinationResolver; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -71,7 +73,7 @@ import org.springframework.util.StringUtils; public class RecipientListRouter extends AbstractMessageRouter implements InitializingBean, RecipientListRouterManagement { - private final ConcurrentLinkedQueue recipients = new ConcurrentLinkedQueue(); + private volatile Queue recipients = new ConcurrentLinkedQueue<>(); /** * Set the channels for this router. Either call this method or @@ -80,12 +82,11 @@ public class RecipientListRouter extends AbstractMessageRouter * @param channels The channels. */ public void setChannels(List channels) { - Assert.notEmpty(channels, "channels must not be empty"); - List recipients = new ArrayList(); - for (MessageChannel channel : channels) { - recipients.add(new Recipient(channel)); - } - this.setRecipients(recipients); + Assert.notEmpty(channels, "'channels' must not be empty"); + List recipients = channels.stream() + .map(Recipient::new) + .collect(Collectors.toList()); + setRecipients(recipients); } /** @@ -93,13 +94,17 @@ public class RecipientListRouter extends AbstractMessageRouter * @param recipients The recipients. */ public void setRecipients(List recipients) { - Assert.notEmpty(recipients, "recipients must not be empty"); - ConcurrentLinkedQueue originalRecipients = this.recipients; - this.recipients.clear(); - this.recipients.addAll(recipients); - if (logger.isDebugEnabled()) { - logger.debug("Channel Recipients:" + originalRecipients + " replaced with:" + this.recipients); + Assert.notEmpty(recipients, "'recipients' must not be empty"); + Queue newRecipients = new ConcurrentLinkedQueue<>(recipients); + + if (getBeanFactory() != null) { + newRecipients.forEach(recipient -> recipient.setChannelResolver(getChannelResolver())); } + if (logger.isDebugEnabled()) { + logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients); + } + + this.recipients = newRecipients; } /** @@ -109,57 +114,72 @@ public class RecipientListRouter extends AbstractMessageRouter @Override @ManagedAttribute public void setRecipientMappings(Map recipientMappings) { - Assert.notEmpty(recipientMappings, "recipientMappings must not be empty"); + Assert.notEmpty(recipientMappings, "'recipientMappings' must not be empty"); Assert.noNullElements(recipientMappings.keySet().toArray(), "'recipientMappings' cannot have null keys."); - ConcurrentLinkedQueue originalRecipients = this.recipients; - this.recipients.clear(); + Queue newRecipients = new ConcurrentLinkedQueue<>(); for (Entry next : recipientMappings.entrySet()) { if (StringUtils.hasText(next.getValue())) { - this.addRecipient(next.getKey(), next.getValue()); + addRecipient(next.getKey(), next.getValue(), newRecipients); } else { - this.addRecipient(next.getKey()); + addRecipient(next.getKey(), (MessageSelector) null, newRecipients); } } if (logger.isDebugEnabled()) { - logger.debug("Channel Recipients:" + originalRecipients + " replaced with:" + this.recipients); + logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients); } - } - @Override - public String getComponentType() { - return "recipient-list-router"; - } - - - @Override - protected Collection determineTargetChannels(Message message) { - List channels = new ArrayList(); - for (Recipient recipient : this.recipients) { - if (recipient.accept(message)) { - channels.add(recipient.getChannel()); - } - } - return channels; + this.recipients = newRecipients; } @Override @ManagedOperation public void addRecipient(String channelName, String selectorExpression) { + addRecipient(channelName, selectorExpression, this.recipients); + } + + private void addRecipient(String channelName, String selectorExpression, Queue recipients) { Assert.hasText(channelName, "'channelName' must not be empty."); Assert.hasText(selectorExpression, "'selectorExpression' must not be empty."); - MessageChannel channel = getChannelResolver().resolveDestination(channelName); - ExpressionEvaluatingSelector expressionEvaluatingSelector = new ExpressionEvaluatingSelector(selectorExpression); - expressionEvaluatingSelector.setBeanFactory(this.getBeanFactory()); - this.recipients.add(new Recipient(channel, expressionEvaluatingSelector)); + ExpressionEvaluatingSelector expressionEvaluatingSelector = + new ExpressionEvaluatingSelector(selectorExpression); + expressionEvaluatingSelector.setBeanFactory(getBeanFactory()); + Recipient recipient = new Recipient(channelName, expressionEvaluatingSelector); + if (getBeanFactory() != null) { + recipient.setChannelResolver(getChannelResolver()); + } + recipients.add(recipient); } @Override @ManagedOperation public void addRecipient(String channelName) { + addRecipient(channelName, (MessageSelector) null); + } + + public void addRecipient(String channelName, MessageSelector selector) { + addRecipient(channelName, selector, this.recipients); + } + + private void addRecipient(String channelName, MessageSelector selector, Queue recipients) { Assert.hasText(channelName, "'channelName' must not be empty."); - MessageChannel channel = getChannelResolver().resolveDestination(channelName); - this.recipients.add(new Recipient(channel)); + Recipient recipient = new Recipient(channelName, selector); + if (getBeanFactory() != null) { + recipient.setChannelResolver(getChannelResolver()); + } + recipients.add(recipient); + } + + public void addRecipient(MessageChannel channel) { + addRecipient(channel, null); + } + + public void addRecipient(MessageChannel channel, MessageSelector selector) { + Recipient recipient = new Recipient(channel, selector); + if (getBeanFactory() != null) { + recipient.setChannelResolver(getChannelResolver()); + } + this.recipients.add(recipient); } @Override @@ -195,18 +215,12 @@ public class RecipientListRouter extends AbstractMessageRouter return counter; } - @Override - @ManagedAttribute - public Collection getRecipients() { - return Collections.unmodifiableCollection(this.recipients); - } - @Override @ManagedOperation public void replaceRecipients(Properties recipientMappings) { Assert.notEmpty(recipientMappings, "'recipientMappings' must not be empty"); Set keys = recipientMappings.stringPropertyNames(); - ConcurrentLinkedQueue originalRecipients = this.recipients; + Queue originalRecipients = this.recipients; this.recipients.clear(); for (String key : keys) { Assert.notNull(key, "channelName can't be null."); @@ -222,13 +236,41 @@ public class RecipientListRouter extends AbstractMessageRouter } } + @Override + @ManagedAttribute + public Collection getRecipients() { + return Collections.unmodifiableCollection(this.recipients); + } + + @Override + public String getComponentType() { + return "recipient-list-router"; + } + + @Override + protected Collection determineTargetChannels(Message message) { + return this.recipients.stream() + .filter(recipient -> recipient.accept(message)) + .map(Recipient::getChannel) + .collect(Collectors.toList()); + } + + + @Override + protected void onInit() throws Exception { + super.onInit(); + this.recipients.forEach(recipient -> recipient.setChannelResolver(getChannelResolver())); + } public static class Recipient { - private final MessageChannel channel; - private final MessageSelector selector; + private MessageChannel channel; + + private String channelName; + + private DestinationResolver channelResolver; public Recipient(MessageChannel channel) { this(channel, null); @@ -239,11 +281,31 @@ public class RecipientListRouter extends AbstractMessageRouter this.selector = selector; } + public Recipient(String channelName) { + this(channelName, null); + } + + public Recipient(String channelName, MessageSelector selector) { + this.channelName = channelName; + this.selector = selector; + } + + public void setChannelResolver(DestinationResolver channelResolver) { + this.channelResolver = channelResolver; + } + private MessageSelector getSelector() { return this.selector; } public MessageChannel getChannel() { + String channelName = this.channelName; + if (channelName != null) { + if (this.channelResolver != null) { + this.channel = this.channelResolver.resolveDestination(channelName); + this.channelName = null; + } + } return this.channel; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java index e768fc5d4d..8b0dd66db6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java @@ -25,6 +25,7 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.handler.GenericHandler; +import org.springframework.integration.handler.LambdaMessageProcessor; import org.springframework.messaging.support.GenericMessage; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 23ccd0dfa5..5696f42835 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -227,7 +227,7 @@ public class AsyncGatewayTests { @Test - public void promiseWithMessageReturned() throws Exception { + public void monoWithMessageReturned() throws Exception { QueueChannel requestChannel = new QueueChannel(); startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); @@ -237,13 +237,13 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Mono> promise = service.returnMessagePromise("foo"); - Object result = promise.block(Duration.ofSeconds(10)); + Mono> mono = service.returnMessagePromise("foo"); + Object result = mono.block(Duration.ofSeconds(10)); assertEquals("foobar", ((Message) result).getPayload()); } @Test - public void promiseWithPayloadReturned() throws Exception { + public void monoWithPayloadReturned() throws Exception { QueueChannel requestChannel = new QueueChannel(); startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); @@ -253,13 +253,13 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Mono promise = service.returnStringPromise("foo"); - Object result = promise.block(Duration.ofSeconds(10)); + Mono mono = service.returnStringPromise("foo"); + Object result = mono.block(Duration.ofSeconds(10)); assertEquals("foobar", result); } @Test - public void promiseWithWildcardReturned() throws Exception { + public void monoWithWildcardReturned() throws Exception { QueueChannel requestChannel = new QueueChannel(); startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); @@ -269,14 +269,14 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Mono promise = service.returnSomethingPromise("foo"); - Object result = promise.block(Duration.ofSeconds(10)); + Mono mono = service.returnSomethingPromise("foo"); + Object result = mono.block(Duration.ofSeconds(10)); assertNotNull(result); assertEquals("foobar", result); } @Test - public void promiseWithConsumer() throws Exception { + public void monoWithConsumer() throws Exception { QueueChannel requestChannel = new QueueChannel(); startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); @@ -286,12 +286,12 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Mono promise = service.returnStringPromise("foo"); + Mono mono = service.returnStringPromise("foo"); final AtomicReference result = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); - promise.subscribe(s -> { + mono.subscribe(s -> { result.set(s); latch.countDown(); }); diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index ec1e500ec4..f4c20d71f7 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -584,28 +584,26 @@ CompletableFuture result = process("foo") String out = result.get(10, TimeUnit.SECONDS); ---- -===== Reactor Promise +===== Reactor Mono -Starting with _version 4.1_, the `GatewayProxyFactoryBean` allows the use of a `Reactor` with gateway interface methods, utilizing a https://github.com/reactor/reactor/wiki/Promises[`Promise`] return type. -The internal `AsyncInvocationTask` is wrapped in a `reactor.function.Supplier`, using a default `RingBufferDispatcher` for the `Promise` consumption. -Only methods with the `Promise` return type are run on the reactor's dispatcher. +Starting with _version 5.0_, the `GatewayProxyFactoryBean` allows the use of the Project Reactor with gateway interface methods, utilizing a https://github.com/reactor/reactor-core[`Mono`] return type. +The internal `AsyncInvocationTask` is wrapped in a `Mono.fromCallable()`. -A `Promise` can be used to retrieve the result later (similar to a `Future`) or you can consume from it with the dispatcher invoking your `Consumer` when the result is returned to the gateway. +A `Mono` can be used to retrieve the result later (similar to a `Future`) or you can consume from it with the dispatcher invoking your `Consumer` when the result is returned to the gateway. -IMPORTANT: The `Promise` isn't _flushed_ immediately by the framework. +IMPORTANT: The `Mono` isn't _flushed_ immediately by the framework. Hence the underlying message flow won't be started before the gateway method returns (as it is with `Future` `Executor` task). -The flow will be started when the `Promise` is _flushed_ or via `Promise.await()`. -Alternatively, the `Promise` (being a `Composable`) might be a part of Reactor `Stream`, when the `flush()` is related to the entire `Stream`. +The flow will be started when the `Mono` is _subscribed_. +Alternatively, the `Mono` (being a `Composable`) might be a part of Reactor stream, when the `subscribe()` is related to the entire `Flux`. For example: - [source,java] ---- @MessagingGateway public static interface TestGateway { @Gateway(requestChannel = "promiseChannel") - Promise multiply(Integer value); + Mono multiply(Integer value); } @@ -618,27 +616,20 @@ public static interface TestGateway { ... - Streams.defer(Arrays.asList("1", "2", "3", "4", "5")) - .get() - .map(Integer::parseInt) - .mapMany(integer -> testGateway.multiply(integer)) - .collect() - .consume(integers -> ...) - .flush(); + Flux.just("1", "2", "3", "4", "5") + .map(Integer::parseInt) + .flatMap(this.testGateway::multiply) + .collectList() + .subscribe(integers -> ...); + ---- Another example is a simple callback scenario: [source,java] ---- -Promise promise = service.process(myOrder); +Mono mono = service.process(myOrder); -promise.consume(new Consumer() { - @Override - public void accept(Invoice invoice) { - handleInvoice(invoice); - } -}) -.flush(); +mono.subscribe(invoice -> handleInvoice(invoice)); ---- The calling thread continues, with `handleInvoice()` being called when the flow completes.