Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove DSL Deprecated Methods and Others

Browse Source 
Since the DSL code is new in this version there is no reason to keep deprecated method

* Remove deprecated `resequence()` and `aggregate()` in the `IntegrationFlowDefinition`
* Remove `InternalAggregatingMessageHandler` with the `MessageGroupProcessorWrapper` delegation logic in favor of newly introduced `AbstractCorrelatingMessageHandler.setOutputProcessor()`
* Rework `Promise` Gateway Docs to the `Mono`
* Rename `AsyncGatewayTests` "promise" words to "mono"
* Resolve `com.rabbitmq.client.FlowListener` deprecation

RecipientListRouter DSL refactoring

Previously there was a DSL specific `DslRecipientListRouter` to overcome the lack of setters in the target `RecipientListRouter` and its `Recipient`

* Add `channelName` variant for the `Recipient`
* Add several new `addRecipient()` methods to the `RecipientListRouter`

All those improvements allow to avoid extra bridge (or adapter) component between Java DSL and target `RecipientListRouter`

* Make `LambdaMessageProcessor` and `GatewayMessageHandler` as `public` classes and move them to the appropriate packages

Address PR comments

Add `isLambda()` condition to `RecipientListRouterSpec` to avoid ambiguity for the provided `GenericSelector` impl

checkstyle polishing
This commit is contained in:
Artem Bilan
2016-11-17 18:56:09 -05:00
committed by Gary Russell
parent b60e1dcb50
commit 092d876fae
16 changed files with 248 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
View File

@@ -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;
}

16
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -93,7 +93,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
private final Map<UUID, ScheduledFuture<?>> 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

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

@@ -38,7 +38,7 @@ import org.springframework.util.Assert;
public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends AbstractMessageRouter>
extends MessageHandlerSpec<S, R> implements ComponentsRegistration {
protected final List<Object> subFlows = new ArrayList<Object>();
protected final List<Object> componentsToRegister = new ArrayList<>();
private boolean defaultToParentFlow;
@@ -102,7 +102,7 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, 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<S extends AbstractRouterSpec<S, R>, R extends Ab
@Override
public Collection<Object> getComponentsToRegister() {
return this.subFlows;
return this.componentsToRegister;
}
}

61
spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java
View File

@@ -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, AggregatingMessageHandler> {
AggregatorSpec() {
super(new InternalAggregatingMessageHandler());
super(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()));
}
/**
@@ -45,7 +40,7 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
* and {@link org.springframework.integration.aggregator.MethodInvokingReleaseStrategy} using the target
* object which should have methods annotated appropriately for each function.
* Also set the output processor.
* @param target the target object.
* @param target the target object.
* @return the handler spec.
*/
public AggregatorSpec processor(Object target) {
@@ -57,17 +52,16 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
* and {@link org.springframework.integration.aggregator.MethodInvokingReleaseStrategy} using the target
* object which should have methods annotated appropriately for each function.
* Also set the output processor.
* @param target the target object.
* @param target the target object.
* @param methodName The method name for the output processor (or 'null' in which case, the
* target object must have an {@link org.springframework.integration.annotation.Aggregator}
* annotation).
* target object must have an {@link org.springframework.integration.annotation.Aggregator} annotation).
* @return the handler spec.
*/
public AggregatorSpec processor(Object target, String methodName) {
super.processor(target);
return this.outputProcessor(methodName != null
? new MethodInvokingMessageGroupProcessor(target, methodName)
: new MethodInvokingMessageGroupProcessor(target));
? new MethodInvokingMessageGroupProcessor(target, methodName)
: new MethodInvokingMessageGroupProcessor(target));
}
/**
@@ -87,8 +81,7 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
* @return the aggregator spec.
*/
public AggregatorSpec outputProcessor(MessageGroupProcessor outputProcessor) {
Assert.notNull(outputProcessor, "'outputProcessor' must not be null.");
((InternalAggregatingMessageHandler) this.handler).getOutputProcessor().setDelegate(outputProcessor);
this.handler.setOutputProcessor(outputProcessor);
return _this();
}
@@ -102,44 +95,4 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
return _this();
}
//TODO Move the logic to the AggregatingMessageHandler
private static class InternalAggregatingMessageHandler extends AggregatingMessageHandler {
InternalAggregatingMessageHandler() {
super(new MessageGroupProcessorWrapper());
}
@Override
protected MessageGroupProcessorWrapper getOutputProcessor() {
return (MessageGroupProcessorWrapper) super.getOutputProcessor();
}
}
private static class MessageGroupProcessorWrapper implements MessageGroupProcessor, BeanFactoryAware {
MessageGroupProcessorWrapper() {
super();
}
private MessageGroupProcessor delegate = new DefaultAggregatingMessageGroupProcessor();
void setDelegate(MessageGroupProcessor delegate) {
this.delegate = delegate;
}
@Override
public Object processMessageGroup(MessageGroup group) {
return this.delegate.processMessageGroup(group);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
if (this.delegate instanceof BeanFactoryAware) {
((BeanFactoryAware) this.delegate).setBeanFactory(beanFactory);
}
}
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java
View File

@@ -49,8 +49,8 @@ import org.springframework.util.Assert;
*
* @since 5.0
*/
public abstract class
CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S, H>, H extends AbstractCorrelatingMessageHandler>
public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S, H>,
H extends AbstractCorrelatingMessageHandler>
extends ConsumerEndpointSpec<S, H> {
private final List<Advice> forceReleaseAdviceChain = new LinkedList<Advice>();
@@ -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)

138
spring-integration-core/src/main/java/org/springframework/integration/dsl/DslRecipientListRouter.java
View File

@@ -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<Tuple2<?, ?>> 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<Recipient> recipients = new ArrayList<Recipient>(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.");
}
}
}
}

1
spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayEndpointSpec.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.dsl;
import org.springframework.integration.gateway.GatewayMessageHandler;
import org.springframework.messaging.MessageChannel;
/**

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

@@ -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 extends IntegrationFlowDefinit
public <S extends AbstractMessageSplitter> B split(S splitter,
Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
Assert.notNull(splitter);
return this.register(new SplitterEndpointSpec<S>(splitter), endpointConfigurer);
return this.register(new SplitterEndpointSpec<>(splitter), endpointConfigurer);
}
/**
@@ -1730,31 +1732,6 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return resequence(null);
}
/**
* Populate the {@link ResequencingMessageHandler} with provided options from {@link ResequencerSpec}.
* In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
* Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"),
* e -> e.phase(100))
* }
* </pre>
* @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<ResequencerSpec> resequencerConfigurer,
Consumer<GenericEndpointSpec<ResequencingMessageHandler>> 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<B extends IntegrationFlowDefinit
return aggregate(null);
}
/**
* Populate the {@link AggregatingMessageHandler} with provided options from {@link AggregatorSpec}.
* In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
* Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .aggregate(a -> a.correlationExpression("1").releaseStrategy(g -> g.size() == 25),
* e -> e.applySequence(false))
* }
* </pre>
* @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<AggregatorSpec> aggregatorConfigurer,
Consumer<GenericEndpointSpec<AggregatingMessageHandler>> 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<B extends IntegrationFlowDefinit
* @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
* @return the current {@link IntegrationFlowDefinition}.
*/
public B route(String beanName, String method, Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
public B route(String beanName, String method,
Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
return route(beanName, method, routerConfigurer, null);
}
@@ -2184,7 +2137,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
public <K, R extends AbstractMappingMessageRouter> B route(R router, Consumer<RouterSpec<K, R>> routerConfigurer,
Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
RouterSpec<K, R> routerSpec = new RouterSpec<K, R>(router);
RouterSpec<K, R> routerSpec = new RouterSpec<>(router);
if (routerConfigurer != null) {
routerConfigurer.accept(routerSpec);
}
@@ -2466,7 +2419,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public <P> B log(Function<Message<P>, Object> function) {
Assert.notNull(function);
return log(new FunctionExpression<Message<P>>(function));
return log(new FunctionExpression<>(function));
}
/**
@@ -2560,7 +2513,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public <P> B log(LoggingHandler.Level level, String category, Function<Message<P>, Object> function) {
Assert.notNull(function);
return log(level, category, new FunctionExpression<Message<P>>(function));
return log(level, category, new FunctionExpression<>(function));
}

66
spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java
View File

@@ -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, RecipientListRouter> {
RecipientListRouterSpec() {
super(new DslRecipientListRouter());
super(new RecipientListRouter());
}
/**
@@ -55,7 +57,13 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @return the router spec.
*/
public RecipientListRouterSpec recipient(String channelName, String expression) {
return recipient(channelName, StringUtils.hasText(expression) ? PARSER.parseExpression(expression) : null);
if (StringUtils.hasText(expression)) {
return recipient(channelName, PARSER.parseExpression(expression));
}
else {
this.target.addRecipient(channelName);
return _this();
}
}
/**
@@ -65,12 +73,12 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @return the router spec.
*/
public RecipientListRouterSpec recipient(String channelName, Expression expression) {
Assert.hasText(channelName);
((DslRecipientListRouter) this.target).add(channelName, expression);
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.target.addRecipient(channelName, selector);
this.componentsToRegister.add(selector);
return _this();
}
/**
* Adds a recipient channel that will be selected if the the selector's accept method returns 'true'.
* @param channelName the channel name.
@@ -78,7 +86,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @return the router spec.
*/
public RecipientListRouterSpec recipientMessageSelector(String channelName, MessageSelector selector) {
return recipient(channelName, (GenericSelector<Message<?>>) selector);
return recipient(channelName, selector);
}
/**
@@ -89,11 +97,24 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @return the router spec.
*/
public <P> RecipientListRouterSpec recipient(String channelName, GenericSelector<P> 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<RecipientListRou
* @return the router spec.
*/
public RecipientListRouterSpec recipient(MessageChannel channel, Expression expression) {
Assert.notNull(channel);
((DslRecipientListRouter) this.target).add(channel, expression);
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.target.addRecipient(channel, selector);
this.componentsToRegister.add(selector);
return _this();
}
@@ -132,7 +154,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @return the router spec.
*/
public RecipientListRouterSpec recipientMessageSelector(MessageChannel channel, MessageSelector selector) {
return recipient(channel, (GenericSelector<Message<?>>) selector);
return recipient(channel, selector);
}
/**
@@ -143,8 +165,14 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @return the router spec.
*/
public <P> RecipientListRouterSpec recipient(MessageChannel channel, GenericSelector<P> 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<RecipientListRou
* @return the router spec.
*/
public RecipientListRouterSpec recipientMessageSelectorFlow(MessageSelector selector, IntegrationFlow subFlow) {
return recipientFlow((GenericSelector<Message<?>>) selector, subFlow);
return recipientFlow(selector, subFlow);
}
/**
@@ -168,8 +196,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
public <P> RecipientListRouterSpec recipientFlow(GenericSelector<P> 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<RecipientListRou
public RecipientListRouterSpec recipientFlow(Expression expression, IntegrationFlow subFlow) {
Assert.notNull(subFlow);
DirectChannel channel = populateSubFlow(subFlow);
((DslRecipientListRouter) this.target).add(channel, expression);
return _this();
return recipient(channel, expression);
}
private DirectChannel populateSubFlow(IntegrationFlow subFlow) {
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
subFlow.configure(flowBuilder);
this.subFlows.add(flowBuilder.get());
this.componentsToRegister.add(flowBuilder.get());
return channel;
}

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

@@ -73,7 +73,7 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
* @see AbstractMappingMessageRouter#setPrefix(String)
*/
public RouterSpec<K, R> 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<K, R extends AbstractMappingMessageRouter>
* @see AbstractMappingMessageRouter#setSuffix(String)
*/
public RouterSpec<K, R> 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<K, R extends AbstractMappingMessageRouter>
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<K, R extends AbstractMappingMessageRouter>
public Collection<Object> 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();
}

14
spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayMessageHandler.java → spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java
View File

@@ -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);
}

13
spring-integration-core/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java → spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java
View File

@@ -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<Object>, BeanFactoryAware {
public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFactoryAware {
private final Object target;
@@ -50,10 +52,9 @@ class LambdaMessageProcessor implements MessageProcessor<Object>, 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<Method> methodValue = new AtomicReference<>();
@@ -79,7 +80,7 @@ class LambdaMessageProcessor implements MessageProcessor<Object>, 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;
}

164
spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -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<Recipient> recipients = new ConcurrentLinkedQueue<Recipient>();
private volatile Queue<Recipient> 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<MessageChannel> channels) {
Assert.notEmpty(channels, "channels must not be empty");
List<Recipient> recipients = new ArrayList<Recipient>();
for (MessageChannel channel : channels) {
recipients.add(new Recipient(channel));
}
this.setRecipients(recipients);
Assert.notEmpty(channels, "'channels' must not be empty");
List<Recipient> 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<Recipient> recipients) {
Assert.notEmpty(recipients, "recipients must not be empty");
ConcurrentLinkedQueue<Recipient> 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<Recipient> 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<String, String> 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<Recipient> originalRecipients = this.recipients;
this.recipients.clear();
Queue<Recipient> newRecipients = new ConcurrentLinkedQueue<>();
for (Entry<String, String> 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<MessageChannel> determineTargetChannels(Message<?> message) {
List<MessageChannel> channels = new ArrayList<MessageChannel>();
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<Recipient> 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<Recipient> 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<Recipient> getRecipients() {
return Collections.unmodifiableCollection(this.recipients);
}
@Override
@ManagedOperation
public void replaceRecipients(Properties recipientMappings) {
Assert.notEmpty(recipientMappings, "'recipientMappings' must not be empty");
Set<String> keys = recipientMappings.stringPropertyNames();
ConcurrentLinkedQueue<Recipient> originalRecipients = this.recipients;
Queue<Recipient> 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<Recipient> getRecipients() {
return Collections.unmodifiableCollection(this.recipients);
}
@Override
public String getComponentType() {
return "recipient-list-router";
}
@Override
protected Collection<MessageChannel> 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<MessageChannel> 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<MessageChannel> 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;
}

1
spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java
View File

@@ -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;

24
spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java
View File

@@ -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<Message<?>> promise = service.returnMessagePromise("foo");
Object result = promise.block(Duration.ofSeconds(10));
Mono<Message<?>> 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<String> promise = service.returnStringPromise("foo");
Object result = promise.block(Duration.ofSeconds(10));
Mono<String> 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<String> promise = service.returnStringPromise("foo");
Mono<String> mono = service.returnStringPromise("foo");
final AtomicReference<String> result = new AtomicReference<String>();
final CountDownLatch latch = new CountDownLatch(1);
promise.subscribe(s -> {
mono.subscribe(s -> {
result.set(s);
latch.countDown();
});

41
src/reference/asciidoc/gateway.adoc
View File

@@ -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<T>`] 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<Integer> multiply(Integer value);
Mono<Integer> 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<Invoice> promise = service.process(myOrder);
Mono<Invoice> mono = service.process(myOrder);
promise.consume(new Consumer<Invoice>() {
@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.