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

GH-3047: Add GatewayProxySpec for Java DSL

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3047

* Improve `GatewayProxyFactoryBean` to determine the return type of the
method call from the interface generic types, when the `serviceInterface`
is a `java.util.function.Function`
* Propagate `MethodArgsHolder` as a `rootObject` for SpEL evaluations
* Deprecate `#gatewayMethod` and `#args` evaluation context variables
in favor of `MethodArgsHolder` as root object.
They will be removed in the future release and a single
`EvaluationContext` will be used for all the gateway expressions
* Introduce an
`IntegrationFlows.from(Class<?> serviceInterface, Consumer<GatewayProxySpec> endpointConfigurer)`
to allow to configure any valid gateway proxy options similar to what
we have with the `<gateway>` and `@MessagingGateway`.
This way we are very close to consistency between different approaches

* * Remove `default` prefix from `GatewayProxySpec` options
* Document the change
This commit is contained in:
Artem Bilan
2019-09-06 13:46:16 -04:00
committed by GitHub
parent c668a046e1
commit 29bebdba97
17 changed files with 565 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java
View File

@@ -102,11 +102,14 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar
BeanDefinitionBuilder.genericBeanDefinition(GatewayMethodMetadata.class);
if (hasDefaultPayloadExpression) {
methodMetadataBuilder.addPropertyValue("payloadExpression", defaultPayloadExpression);
methodMetadataBuilder.addPropertyValue("payloadExpression",
BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class)
.addConstructorArgValue(defaultPayloadExpression)
.getBeanDefinition());
}
if (hasDefaultHeaders) {
Map<String, Object> headerExpressions = new ManagedMap<String, Object>();
Map<String, Object> headerExpressions = new ManagedMap<>();
for (Map<String, Object> header : defaultHeaders) {
String headerValue = (String) header.get("value");
String headerExpression = (String) header.get("expression");
@@ -181,9 +184,11 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar
* @param importingClassMetadata The importing class metadata
* @return The captured values.
*/
private List<MultiValueMap<String, Object>> captureMetaAnnotationValues(AnnotationMetadata importingClassMetadata) {
private static List<MultiValueMap<String, Object>> captureMetaAnnotationValues(
AnnotationMetadata importingClassMetadata) {
Set<String> directAnnotations = importingClassMetadata.getAnnotationTypes();
List<MultiValueMap<String, Object>> valuesHierarchy = new ArrayList<MultiValueMap<String, Object>>();
List<MultiValueMap<String, Object>> valuesHierarchy = new ArrayList<>();
// Need to grab the values now; see SPR-11710
for (String ann : directAnnotations) {
Set<String> chain = importingClassMetadata.getMetaAnnotationTypes(ann);
@@ -203,8 +208,9 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar
* @param valuesHierarchy The values hierarchy in order.
* @param annotationAttributes The current attribute values.
*/
private void replaceEmptyOverrides(List<MultiValueMap<String, Object>> valuesHierarchy,
private static void replaceEmptyOverrides(List<MultiValueMap<String, Object>> valuesHierarchy,
Map<String, Object> annotationAttributes) {
for (Entry<String, Object> entry : annotationAttributes.entrySet()) {
Object value = entry.getValue();
if (!MessagingAnnotationUtils.hasValue(value)) {

27
spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java
View File

@@ -31,6 +31,7 @@ import org.springframework.beans.factory.support.ManagedMap;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.MessagingGatewayRegistrar;
import org.springframework.integration.gateway.GatewayMethodMetadata;
import org.springframework.util.Assert;
@@ -54,7 +55,7 @@ public class GatewayParser implements BeanDefinitionParser {
public BeanDefinition parse(final Element element, ParserContext parserContext) {
boolean isNested = parserContext.isNested();
final Map<String, Object> gatewayAttributes = new HashMap<String, Object>();
final Map<String, Object> gatewayAttributes = new HashMap<>();
gatewayAttributes.put(AbstractBeanDefinitionParser.NAME_ATTRIBUTE,
element.getAttribute(AbstractBeanDefinitionParser.ID_ATTRIBUTE));
gatewayAttributes.put("defaultPayloadExpression", element.getAttribute("default-payload-expression"));
@@ -95,20 +96,19 @@ public class GatewayParser implements BeanDefinitionParser {
}
}
@SuppressWarnings("rawtypes")
private void headers(final Element element, final Map<String, Object> gatewayAttributes) {
private void headers(Element element, Map<String, Object> gatewayAttributes) {
List<Element> headerElements = DomUtils.getChildElementsByTagName(element, "default-header");
if (!CollectionUtils.isEmpty(headerElements)) {
List<Map<String, Object>> headers = new ArrayList<Map<String, Object>>(headerElements.size());
List<Map<String, Object>> headers = new ArrayList<>(headerElements.size());
for (Element e : headerElements) {
Map<String, Object> header = new HashMap<String, Object>();
Map<String, Object> header = new HashMap<>();
header.put(AbstractBeanDefinitionParser.NAME_ATTRIBUTE,
e.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE));
header.put("value", e.getAttribute("value"));
header.put("expression", e.getAttribute("expression"));
headers.add(header);
}
gatewayAttributes.put("defaultHeaders", headers.toArray(new Map[0]));
gatewayAttributes.put("defaultHeaders", headers.toArray(new Map<?, ?>[0]));
}
}
@@ -116,7 +116,7 @@ public class GatewayParser implements BeanDefinitionParser {
final Map<String, Object> gatewayAttributes) {
List<Element> methodElements = DomUtils.getChildElementsByTagName(element, "method");
if (!CollectionUtils.isEmpty(methodElements)) {
Map<String, BeanDefinition> methodMetadataMap = new ManagedMap<String, BeanDefinition>();
Map<String, BeanDefinition> methodMetadataMap = new ManagedMap<>();
for (Element methodElement : methodElements) {
String methodName = methodElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
BeanDefinitionBuilder methodMetadataBuilder = BeanDefinitionBuilder.genericBeanDefinition(
@@ -128,17 +128,22 @@ public class GatewayParser implements BeanDefinitionParser {
methodMetadataBuilder.addPropertyValue("replyTimeout", methodElement.getAttribute("reply-timeout"));
boolean hasMapper = StringUtils.hasText(element.getAttribute("mapper"));
Assert.state(!hasMapper || !StringUtils.hasText(element.getAttribute("payload-expression")),
String payloadExpression = methodElement.getAttribute("payload-expression");
Assert.state(!hasMapper || !StringUtils.hasText(payloadExpression),
"'payload-expression' is not allowed when a 'mapper' is provided");
IntegrationNamespaceUtils.setValueIfAttributeDefined(methodMetadataBuilder, methodElement,
"payload-expression");
if (StringUtils.hasText(payloadExpression)) {
methodMetadataBuilder.addPropertyValue("payloadExpression",
BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class)
.addConstructorArgValue(payloadExpression)
.getBeanDefinition());
}
List<Element> invocationHeaders = DomUtils.getChildElementsByTagName(methodElement, "header");
if (!CollectionUtils.isEmpty(invocationHeaders)) {
Assert.state(!hasMapper, "header elements are not allowed when a 'mapper' is provided");
Map<String, Object> headerExpressions = new ManagedMap<String, Object>();
Map<String, Object> headerExpressions = new ManagedMap<>();
for (Element headerElement : invocationHeaders) {
BeanDefinition expressionDef = IntegrationNamespaceUtils
.createExpressionDefinitionFromValueOrExpression("value", "expression", parserContext,

279
spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java Normal file
View File

@@ -0,0 +1,279 @@
/*
* Copyright 2019 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
*
* https://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.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.function.Function;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.annotation.AnnotationConstants;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean;
import org.springframework.integration.gateway.GatewayMethodMetadata;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.gateway.MethodArgsHolder;
import org.springframework.integration.gateway.MethodArgsMessageMapper;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
/**
* A builder for the {@link GatewayProxyFactoryBean} options
* when {@link org.springframework.integration.annotation.MessagingGateway} on the service interface cannot be
* declared.
*
* @author Artem Bilan
*
* @since 5.2
*/
public class GatewayProxySpec {
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
private final MessageChannel gatewayRequestChannel = new DirectChannel();
private final GatewayProxyFactoryBean gatewayProxyFactoryBean;
private final GatewayMethodMetadata gatewayMethodMetadata = new GatewayMethodMetadata();
private final Map<String, Expression> headerExpressions = new HashMap<>();
private boolean populateGatewayMethodMetadata;
GatewayProxySpec(Class<?> serviceInterface) {
this.gatewayProxyFactoryBean = new AnnotationGatewayProxyFactoryBean(serviceInterface);
this.gatewayProxyFactoryBean.setDefaultRequestChannel(this.gatewayRequestChannel);
}
/**
* Specify a bean name for the target {@link GatewayProxyFactoryBean}.
* @param beanName the bean name to be used for registering bean for the gateway proxy
* @return current {@link GatewayProxySpec}.
*/
public GatewayProxySpec beanName(@Nullable String beanName) {
if (beanName != null) {
this.gatewayProxyFactoryBean.setBeanName(beanName);
}
return this;
}
/**
* Identifies the default channel the gateway proxy will subscribe to, to receive reply
* {@code Message}s, the payloads of
* which will be converted to the return type of the method signature.
* @param channelName the bean name for {@link MessageChannel}
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setDefaultReplyChannel
*/
public GatewayProxySpec replyChannel(String channelName) {
this.gatewayProxyFactoryBean.setDefaultReplyChannelName(channelName);
return this;
}
/**
* Identifies the default channel the gateway proxy will subscribe to, to receive reply
* {@code Message}s, the payloads of
* which will be converted to the return type of the method signature.
* @param replyChannel the {@link MessageChannel} for replies.
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setDefaultReplyChannel
*/
public GatewayProxySpec replyChannel(MessageChannel replyChannel) {
this.gatewayProxyFactoryBean.setDefaultReplyChannel(replyChannel);
return this;
}
/**
* Identifies a channel that error messages will be sent to if a failure occurs in the
* gateway's proxy invocation. If no {@code errorChannel} reference is provided, the gateway will
* propagate {@code Exception}s to the caller. To completely suppress {@code Exception}s, provide a
* reference to the {@code nullChannel} here.
* @param errorChannelName the bean name for {@link MessageChannel}
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setErrorChannel
*/
public GatewayProxySpec errorChannel(String errorChannelName) {
this.gatewayProxyFactoryBean.setErrorChannelName(errorChannelName);
return this;
}
/**
* Identifies a channel that error messages will be sent to if a failure occurs in the
* gateway's proxy invocation. If no {@code errorChannel} reference is provided, the gateway will
* propagate {@code Exception}s to the caller. To completely suppress {@code Exception}s, provide a
* reference to the {@code nullChannel} here.
* @param errorChannel the {@link MessageChannel} for replies.
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setErrorChannel
*/
public GatewayProxySpec errorChannel(MessageChannel errorChannel) {
this.gatewayProxyFactoryBean.setErrorChannel(errorChannel);
return this;
}
/**
* Provides the amount of time dispatcher would wait to send a {@code Message}. This
* timeout would only apply if there is a potential to block in the send call. For
* example if this gateway is hooked up to a {@code QueueChannel}. Value is specified
* in milliseconds.
* @param requestTimeout the timeout for requests in milliseconds.
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setDefaultRequestTimeout
*/
public GatewayProxySpec requestTimeout(long requestTimeout) {
this.gatewayProxyFactoryBean.setDefaultRequestTimeout(requestTimeout);
return this;
}
/**
* Allows to specify how long this gateway will wait for the reply {@code Message}
* before returning. By default it will wait indefinitely. {@code null} is returned if
* the gateway times out. Value is specified in milliseconds.
* @param replyTimeout the timeout for replies in milliseconds.
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setDefaultReplyTimeout
*/
public GatewayProxySpec replyTimeout(long replyTimeout) {
this.gatewayProxyFactoryBean.setDefaultReplyTimeout(replyTimeout);
return this;
}
/**
* Provide a reference to an implementation of {@link Executor}
* to use for any of the interface methods that have a {@link java.util.concurrent.Future} return type.
* This {@code Executor} will only be used for those async methods; the sync methods
* will be invoked in the caller's thread.
* Use {@link AnnotationConstants#NULL} to specify no async executor - for example
* if your downstream flow returns a {@link java.util.concurrent.Future}.
* @param executor the {@link Executor} to use.
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setAsyncExecutor
*/
public GatewayProxySpec asyncExecutor(@Nullable Executor executor) {
this.gatewayProxyFactoryBean.setAsyncExecutor(executor);
return this;
}
/**
* An expression that will be used to generate the {@code payload} for all methods in the service interface
* unless explicitly overridden by a method declaration.
* The root object for evaluation context is {@link MethodArgsHolder}.
* @param expression the SpEL expression for default payload.
* @return current {@link GatewayProxySpec}.
* @see org.springframework.integration.annotation.MessagingGateway#defaultPayloadExpression
*/
public GatewayProxySpec payloadExpression(String expression) {
return payloadExpression(PARSER.parseExpression(expression));
}
/**
* A {@link Function} that will be used to generate the {@code payload} for all methods in the service interface
* unless explicitly overridden by a method declaration.
* @param defaultPayloadFunction the {@link Function} for default payload.
* @return current {@link GatewayProxySpec}.
* @see org.springframework.integration.annotation.MessagingGateway#defaultPayloadExpression
*/
public GatewayProxySpec payloadFunction(Function<MethodArgsHolder, ?> defaultPayloadFunction) {
return payloadExpression(new FunctionExpression<>(defaultPayloadFunction));
}
/**
* An expression that will be used to generate the {@code payload} for all methods in the service interface
* unless explicitly overridden by a method declaration.
* The root object for evaluation context is {@link MethodArgsHolder}.
* a bean resolver is also available, enabling expressions like {@code @someBean(#args)}.
* @param expression the SpEL expression for default payload.
* @return current {@link GatewayProxySpec}.
* @see org.springframework.integration.annotation.MessagingGateway#defaultPayloadExpression
*/
public GatewayProxySpec payloadExpression(Expression expression) {
this.gatewayMethodMetadata.setPayloadExpression(expression);
this.populateGatewayMethodMetadata = true;
return this;
}
/**
* Provides custom message header. The default headers are created for
* all methods on the service-interface (unless overridden by a specific method).
* @param headerName the name ofr the header.
* @param value the static value for the header.
* @return current {@link GatewayProxySpec}.
* @see org.springframework.integration.annotation.MessagingGateway#defaultHeaders
*/
public GatewayProxySpec header(String headerName, Object value) {
return header(headerName, new ValueExpression<>(value));
}
/**
* Provides custom message header. The default headers are created for
* all methods on the service-interface (unless overridden by a specific method).
* @param headerName the name ofr the header.
* @param valueFunction the {@link Function} for the header value.
* @return current {@link GatewayProxySpec}.
* @see org.springframework.integration.annotation.MessagingGateway#defaultHeaders
*/
public GatewayProxySpec header(String headerName, Function<MethodArgsHolder, ?> valueFunction) {
return header(headerName, new FunctionExpression<>(valueFunction));
}
/**
* Provides custom message header. The default headers are created for
* all methods on the service-interface (unless overridden by a specific method).
* This expression-based header can get access to the {@link MethodArgsHolder}
* as a root object for evaluation context.
* @param headerName the name ofr the header.
* @param valueExpression the SpEL expression for the header value.
* @return current {@link GatewayProxySpec}.
* @see org.springframework.integration.annotation.MessagingGateway#defaultHeaders
*/
public GatewayProxySpec header(String headerName, Expression valueExpression) {
this.headerExpressions.put(headerName, valueExpression);
this.populateGatewayMethodMetadata = true;
return this;
}
/**
* An {@link MethodArgsMessageMapper}
* to map the method arguments to a {@link org.springframework.messaging.Message}. When this
* is provided, no {@code payload-expression}s or {@code header}s are allowed; the custom mapper is
* responsible for creating the message.
* @param mapper the {@link MethodArgsMessageMapper} to use.
* @return current {@link GatewayProxySpec}.
* @see GatewayProxyFactoryBean#setMapper(MethodArgsMessageMapper)
*/
public GatewayProxySpec mapper(MethodArgsMessageMapper mapper) {
this.gatewayProxyFactoryBean.setMapper(mapper);
return this;
}
MessageChannel getGatewayRequestChannel() {
return this.gatewayRequestChannel;
}
GatewayProxyFactoryBean getGatewayProxyFactoryBean() {
if (this.populateGatewayMethodMetadata) {
this.gatewayMethodMetadata.setHeaderExpressions(this.headerExpressions);
this.gatewayProxyFactoryBean.setGlobalMethodMetadata(this.gatewayMethodMetadata);
}
return this.gatewayProxyFactoryBean;
}
}

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

@@ -26,6 +26,7 @@ import org.springframework.context.SmartLifecycle;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -140,11 +141,13 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
protected IntegrationFlowDefinition<?> from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.from(messageSourceSpec, endpointConfigurer);
}
protected IntegrationFlowDefinition<?> from(MessageSource<?> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.from(messageSource, endpointConfigurer);
}
@@ -182,6 +185,7 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
protected IntegrationFlowBuilder from(Object service, String methodName,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.from(service, methodName, endpointConfigurer);
}
@@ -191,6 +195,7 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
protected <T> IntegrationFlowBuilder from(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.from(messageSource, endpointConfigurer);
}
@@ -198,8 +203,29 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
return IntegrationFlows.from(serviceInterface);
}
protected IntegrationFlowBuilder from(Class<?> serviceInterface, String beanName) {
return IntegrationFlows.from(serviceInterface, beanName);
/**
* Start a flow from a proxy for the service interface.
* @param serviceInterface the service interface to proxy for the gateway.
* @param beanName the bean name for the gateway proxy.
* @return the {@link IntegrationFlowBuilder} instance
* @deprecated since 5.2 in favor of {@link #from(Class, Consumer)}
*/
@Deprecated
protected IntegrationFlowBuilder from(Class<?> serviceInterface, @Nullable String beanName) {
return from(serviceInterface, (gateway) -> gateway.beanName(beanName));
}
/**
* Start a flow from a proxy for the service interface.
* @param serviceInterface the service interface class.
* @param endpointConfigurer the {@link Consumer} to configure proxy bean for gateway.
* @return new {@link IntegrationFlowBuilder}.
* @since 5.2
*/
protected IntegrationFlowBuilder from(Class<?> serviceInterface,
@Nullable Consumer<GatewayProxySpec> endpointConfigurer) {
return IntegrationFlows.from(serviceInterface, endpointConfigurer);
}
protected IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {

35
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java
View File

@@ -28,8 +28,6 @@ import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototy
import org.springframework.integration.dsl.support.MessageChannelReference;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -316,7 +314,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(Class<?> serviceInterface) {
return from(serviceInterface, null);
return from(serviceInterface, (Consumer<GatewayProxySpec>) null);
}
/**
@@ -331,21 +329,38 @@ public final class IntegrationFlows {
* {@link org.springframework.integration.annotation.MessagingGateway} annotation.
* @param beanName the bean name to be used for registering bean for the gateway proxy
* @return new {@link IntegrationFlowBuilder}.
* @deprecated since 5.2 in favor of {@link #from(Class, Consumer)}
*/
@Deprecated
public static IntegrationFlowBuilder from(Class<?> serviceInterface, @Nullable String beanName) {
final DirectChannel gatewayRequestChannel = new DirectChannel();
return from(serviceInterface, gateway -> gateway.beanName(beanName));
}
GatewayProxyFactoryBean gatewayProxyFactoryBean = new AnnotationGatewayProxyFactoryBean(serviceInterface);
/**
* Populate the {@link MessageChannel} to the new {@link IntegrationFlowBuilder}
* chain, which becomes as a {@code requestChannel} for the Messaging Gateway(s) built
* on the provided service interface.
* <p>A gateway proxy bean for provided service interface is based on the options
* configured via provided {@link Consumer}.
* @param serviceInterface the service interface class with an optional
* {@link org.springframework.integration.annotation.MessagingGateway} annotation.
* @param endpointConfigurer the {@link Consumer} to configure proxy bean for gateway.
* @return new {@link IntegrationFlowBuilder}.
* @since 5.2
*/
public static IntegrationFlowBuilder from(Class<?> serviceInterface,
@Nullable Consumer<GatewayProxySpec> endpointConfigurer) {
gatewayProxyFactoryBean.setDefaultRequestChannel(gatewayRequestChannel);
if (beanName != null) {
gatewayProxyFactoryBean.setBeanName(beanName);
GatewayProxySpec gatewayProxySpec = new GatewayProxySpec(serviceInterface);
if (endpointConfigurer != null) {
endpointConfigurer.accept(gatewayProxySpec);
}
return from(gatewayRequestChannel)
.addComponent(gatewayProxyFactoryBean);
return from(gatewayProxySpec.getGatewayRequestChannel())
.addComponent(gatewayProxySpec.getGatewayProxyFactoryBean());
}
/**
* Populate a {@link FluxMessageChannel} to the {@link IntegrationFlowBuilder} chain
* and subscribe it to the provided {@link Publisher}.

2
spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java
View File

@@ -121,7 +121,7 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean {
GatewayMethodMetadata gatewayMethodMetadata = new GatewayMethodMetadata();
if (hasDefaultPayloadExpression) {
gatewayMethodMetadata.setPayloadExpression(defaultPayloadExpression);
gatewayMethodMetadata.setPayloadExpression(EXPRESSION_PARSER.parseExpression(defaultPayloadExpression));
}
Map<String, Expression> headerExpressions = Arrays.stream(defaultHeaders)

32
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java
View File

@@ -161,8 +161,8 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
}
public void setPayloadExpression(String expressionString) {
this.payloadExpression = PARSER.parseExpression(expressionString);
public void setPayloadExpression(Expression expressionString) {
this.payloadExpression = expressionString;
}
@Override
@@ -204,20 +204,20 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
}
private Map<String, Object> evaluateHeaders(EvaluationContext methodInvocationEvaluationContext,
Map<String, Expression> headerExpressions) {
MethodArgsHolder methodArgsHolder, Map<String, Expression> headerExpressions) {
Map<String, Object> evaluatedHeaders = new HashMap<>();
for (Map.Entry<String, Expression> entry : headerExpressions.entrySet()) {
Object value = entry.getValue().getValue(methodInvocationEvaluationContext);
Object value = entry.getValue().getValue(methodInvocationEvaluationContext, methodArgsHolder);
evaluatedHeaders.put(entry.getKey(), value);
}
return evaluatedHeaders;
}
// TODO Remove in the future release. The MethodArgsHolder as a root object covers this use-case.
private StandardEvaluationContext createMethodInvocationEvaluationContext(Object[] arguments) {
StandardEvaluationContext context = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
context.setVariable("args", arguments);
context.setVariable("gatewayMethod", this.method);
return context;
}
@@ -301,7 +301,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
if (GatewayMethodInboundMessageMapper.this.payloadExpression != null) {
messageOrPayload =
GatewayMethodInboundMessageMapper.this.payloadExpression.getValue(
methodInvocationEvaluationContext);
methodInvocationEvaluationContext, holder);
}
for (int i = 0; i < GatewayMethodInboundMessageMapper.this.parameterList.size(); i++) {
Object argumentValue = arguments[i];
@@ -334,13 +334,13 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
Assert.isTrue(messageOrPayload != null,
() -> "unable to determine a Message or payload parameter on method ["
+ GatewayMethodInboundMessageMapper.this.method + "]");
populateSendAndReplyTimeoutHeaders(methodInvocationEvaluationContext, headersToPopulate);
return buildMessage(headersToPopulate, messageOrPayload, methodInvocationEvaluationContext);
populateSendAndReplyTimeoutHeaders(methodInvocationEvaluationContext, holder, headersToPopulate);
return buildMessage(holder, headersToPopulate, messageOrPayload, methodInvocationEvaluationContext);
}
private void headerOrHeaders(Map<String, Object> headersToPopulate, Object argumentValue,
MethodParameter methodParameter, Annotation annotation) {
if (annotation.annotationType().equals(Header.class)) {
processHeaderAnnotation(headersToPopulate, argumentValue, methodParameter, annotation);
}
@@ -405,22 +405,22 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
}
private void populateSendAndReplyTimeoutHeaders(EvaluationContext methodInvocationEvaluationContext,
Map<String, Object> headersToPopulate) {
MethodArgsHolder methodArgsHolder, Map<String, Object> headersToPopulate) {
if (GatewayMethodInboundMessageMapper.this.sendTimeoutExpression != null) {
headersToPopulate.computeIfAbsent(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER,
v -> GatewayMethodInboundMessageMapper.this.sendTimeoutExpression
.getValue(methodInvocationEvaluationContext, Long.class));
.getValue(methodInvocationEvaluationContext, methodArgsHolder, Long.class));
}
if (GatewayMethodInboundMessageMapper.this.replyTimeoutExpression != null) {
headersToPopulate.computeIfAbsent(GenericMessagingTemplate.DEFAULT_RECEIVE_TIMEOUT_HEADER,
v -> GatewayMethodInboundMessageMapper.this.replyTimeoutExpression
.getValue(methodInvocationEvaluationContext, Long.class));
.getValue(methodInvocationEvaluationContext, methodArgsHolder, Long.class));
}
}
private Message<?> buildMessage(Map<String, Object> headers, Object messageOrPayload,
EvaluationContext methodInvocationEvaluationContext) {
private Message<?> buildMessage(MethodArgsHolder methodArgsHolder, Map<String, Object> headers,
Object messageOrPayload, EvaluationContext methodInvocationEvaluationContext) {
AbstractIntegrationMessageBuilder<?> builder =
(messageOrPayload instanceof Message)
@@ -430,13 +430,13 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
// Explicit headers in XML override any @Header annotations...
if (!CollectionUtils.isEmpty(GatewayMethodInboundMessageMapper.this.headerExpressions)) {
Map<String, Object> evaluatedHeaders = evaluateHeaders(methodInvocationEvaluationContext,
GatewayMethodInboundMessageMapper.this.headerExpressions);
methodArgsHolder, GatewayMethodInboundMessageMapper.this.headerExpressions);
builder.copyHeaders(evaluatedHeaders);
}
// ...whereas global (default) headers do not...
if (!CollectionUtils.isEmpty(GatewayMethodInboundMessageMapper.this.globalHeaderExpressions)) {
Map<String, Object> evaluatedHeaders = evaluateHeaders(methodInvocationEvaluationContext,
GatewayMethodInboundMessageMapper.this.globalHeaderExpressions);
methodArgsHolder, GatewayMethodInboundMessageMapper.this.globalHeaderExpressions);
builder.copyHeadersIfAbsent(evaluatedHeaders);
}
if (GatewayMethodInboundMessageMapper.this.headers != null) {

28
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodMetadata.java
View File

@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import org.springframework.expression.Expression;
import org.springframework.lang.Nullable;
/**
* Represents the metadata associated with a Gateway method. This is most useful when there are
@@ -30,28 +31,30 @@ import org.springframework.expression.Expression;
*
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class GatewayMethodMetadata {
private volatile String payloadExpression;
private final Map<String, Expression> headerExpressions = new HashMap<>();
private volatile String requestChannelName;
private Expression payloadExpression;
private volatile String replyChannelName;
private String requestChannelName;
private volatile String requestTimeout;
private String replyChannelName;
private volatile String replyTimeout;
private String requestTimeout;
private volatile Map<String, Expression> headerExpressions = new HashMap<String, Expression>();
private String replyTimeout;
public String getPayloadExpression() {
@Nullable
public Expression getPayloadExpression() {
return this.payloadExpression;
}
public void setPayloadExpression(String payloadExpression) {
public void setPayloadExpression(Expression payloadExpression) {
this.payloadExpression = payloadExpression;
}
@@ -59,8 +62,11 @@ public class GatewayMethodMetadata {
return this.headerExpressions;
}
public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
this.headerExpressions = headerExpressions;
public void setHeaderExpressions(@Nullable Map<String, Expression> headerExpressions) {
this.headerExpressions.clear();
if (headerExpressions != null) {
this.headerExpressions.putAll(headerExpressions);
}
}
public String getRequestChannelName() {

136
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -31,6 +31,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.aopalliance.intercept.MethodInterceptor;
@@ -46,6 +47,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -53,6 +55,7 @@ import org.springframework.core.task.support.TaskExecutorAdapter;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.GatewayHeader;
@@ -100,51 +103,53 @@ import reactor.core.publisher.Mono;
public class GatewayProxyFactoryBean extends AbstractEndpoint
implements TrackableComponent, FactoryBean<Object>, MethodInterceptor, BeanClassLoaderAware {
private volatile Class<?> serviceInterface;
private volatile MessageChannel defaultRequestChannel;
private volatile String defaultRequestChannelName;
private volatile MessageChannel defaultReplyChannel;
private volatile String defaultReplyChannelName;
private volatile MessageChannel errorChannel;
private volatile String errorChannelName;
private volatile Expression defaultRequestTimeout;
private volatile Expression defaultReplyTimeout;
private volatile DestinationResolver<MessageChannel> channelResolver;
private volatile boolean shouldTrack = false;
private volatile TypeConverter typeConverter = new SimpleTypeConverter();
private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
private volatile Object serviceProxy;
private final Map<Method, MethodInvocationGateway> gatewayMap = new HashMap<>();
private volatile AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor();
private volatile Class<?> asyncSubmitType;
private volatile Class<?> asyncSubmitListenableType;
private volatile boolean initialized;
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
private final Object initializationMonitor = new Object();
private volatile Map<String, GatewayMethodMetadata> methodMetadataMap;
private final Map<Method, MethodInvocationGateway> gatewayMap = new HashMap<>();
private volatile GatewayMethodMetadata globalMethodMetadata;
private Class<?> serviceInterface;
private volatile MethodArgsMessageMapper argsMapper;
private MessageChannel defaultRequestChannel;
private String defaultRequestChannelName;
private MessageChannel defaultReplyChannel;
private String defaultReplyChannelName;
private MessageChannel errorChannel;
private String errorChannelName;
private Expression defaultRequestTimeout;
private Expression defaultReplyTimeout;
private DestinationResolver<MessageChannel> channelResolver;
private boolean shouldTrack = false;
private TypeConverter typeConverter = new SimpleTypeConverter();
private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
private Object serviceProxy;
private AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor();
private Class<?> asyncSubmitType;
private Class<?> asyncSubmitListenableType;
private volatile boolean initialized;
private Map<String, GatewayMethodMetadata> methodMetadataMap;
private GatewayMethodMetadata globalMethodMetadata;
private MethodArgsMessageMapper argsMapper;
private EvaluationContext evaluationContext = new StandardEvaluationContext();
@@ -484,13 +489,17 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
@Nullable
private Object invokeGatewayMethod(MethodInvocation invocation, boolean runningOnCallerThread) {
if (!this.initialized) {
this.afterPropertiesSet();
afterPropertiesSet();
}
Method method = invocation.getMethod();
MethodInvocationGateway gateway = this.gatewayMap.get(method);
Class<?> returnType = method.getReturnType();
boolean shouldReturnMessage = Message.class.isAssignableFrom(returnType)
|| hasReturnParameterizedWithMessage(method, runningOnCallerThread);
if (gateway.isReturnTypeMessage == null) {
gateway.isReturnTypeMessage =
Message.class.isAssignableFrom(returnType) || hasReturnMessageTypeOnFunction(method);
}
boolean shouldReturnMessage =
gateway.isReturnTypeMessage || hasReturnParameterizedWithMessage(method, runningOnCallerThread);
boolean shouldReply = returnType != void.class;
int paramCount = method.getParameterTypes().length;
Object response = null;
@@ -539,10 +548,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
// check for the method metadata next
if (this.methodMetadataMap != null) {
GatewayMethodMetadata metadata = this.methodMetadataMap.get(method.getName());
hasPayloadExpression = (metadata != null) && StringUtils.hasText(metadata.getPayloadExpression());
hasPayloadExpression = (metadata != null) && metadata.getPayloadExpression() != null;
}
else if (this.globalMethodMetadata != null) {
hasPayloadExpression = StringUtils.hasText(this.globalMethodMetadata.getPayloadExpression());
hasPayloadExpression = this.globalMethodMetadata.getPayloadExpression() != null;
}
}
return hasPayloadExpression;
@@ -563,7 +572,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
return response;
}
private void rethrowExceptionCauseIfPossible(Throwable originalException, Method method) throws Throwable { // NOSONAR
private void rethrowExceptionCauseIfPossible(Throwable originalException, Method method)
throws Throwable { // NOSONAR
Class<?>[] exceptionTypes = method.getExceptionTypes();
Throwable t = originalException;
while (t != null) {
@@ -589,10 +599,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
String replyChannelName = null;
Expression requestTimeout = this.defaultRequestTimeout;
Expression replyTimeout = this.defaultReplyTimeout;
String payloadExpression = this.globalMethodMetadata != null
Expression payloadExpression = this.globalMethodMetadata != null
? this.globalMethodMetadata.getPayloadExpression()
: null;
Map<String, Expression> headerExpressions = new HashMap<String, Expression>();
Map<String, Expression> headerExpressions = new HashMap<>();
if (gatewayAnnotation != null) {
requestChannelName = gatewayAnnotation.requestChannel();
replyChannelName = gatewayAnnotation.replyChannel();
@@ -615,8 +625,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
if (StringUtils.hasText(gatewayAnnotation.replyTimeoutExpression())) {
replyTimeout = ExpressionUtils.longExpression(gatewayAnnotation.replyTimeoutExpression());
}
if (payloadExpression == null || StringUtils.hasText(gatewayAnnotation.payloadExpression())) {
payloadExpression = gatewayAnnotation.payloadExpression();
if (payloadExpression == null && StringUtils.hasText(gatewayAnnotation.payloadExpression())) {
payloadExpression = PARSER.parseExpression(gatewayAnnotation.payloadExpression());
}
annotationHeaders(gatewayAnnotation, headerExpressions);
@@ -624,7 +634,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
else if (this.methodMetadataMap != null && this.methodMetadataMap.size() > 0) {
GatewayMethodMetadata methodMetadata = this.methodMetadataMap.get(method.getName());
if (methodMetadata != null) {
if (StringUtils.hasText(methodMetadata.getPayloadExpression())) {
if (methodMetadata.getPayloadExpression() != null) {
payloadExpression = methodMetadata.getPayloadExpression();
}
if (!CollectionUtils.isEmpty(methodMetadata.getHeaderExpressions())) {
@@ -647,13 +657,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
GatewayMethodInboundMessageMapper messageMapper = new GatewayMethodInboundMessageMapper(method,
headerExpressions,
this.globalMethodMetadata != null ? this.globalMethodMetadata.getHeaderExpressions() : null,
headers, this.argsMapper, this.getMessageBuilderFactory());
headers, this.argsMapper, getMessageBuilderFactory());
MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper);
JavaUtils.INSTANCE
.acceptIfHasText(payloadExpression, messageMapper::setPayloadExpression)
.acceptIfNotNull(getTaskScheduler(), gateway::setTaskScheduler);
gateway.setBeanName(this.getComponentName());
.acceptIfNotNull(payloadExpression, messageMapper::setPayloadExpression)
.acceptIfNotNull(getTaskScheduler(), gateway::setTaskScheduler);
gateway.setBeanName(getComponentName());
setChannel(this.errorChannel, gateway::setErrorChannel, this.errorChannelName, gateway::setErrorChannelName);
setChannel(requestChannelName, this.defaultRequestChannelName, gateway::setRequestChannelName,
@@ -662,7 +673,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
this.defaultReplyChannel, gateway::setReplyChannel);
timeouts(requestTimeout, replyTimeout, messageMapper, gateway);
BeanFactory beanFactory = this.getBeanFactory();
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
gateway.setBeanFactory(beanFactory);
messageMapper.setBeanFactory(beanFactory);
@@ -842,11 +853,24 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
return false;
}
private boolean hasReturnMessageTypeOnFunction(Method method) {
if (Function.class.isAssignableFrom(this.serviceInterface) && "apply".equals(method.getName())) {
Class<?> returnType =
ResolvableType.forClass(Function.class, this.serviceInterface)
.getGeneric(1)
.getRawClass();
return returnType != null && Message.class.isAssignableFrom(returnType);
}
return false;
}
private static final class MethodInvocationGateway extends MessagingGatewaySupport {
private Expression receiveTimeoutExpression;
volatile Boolean isReturnTypeMessage;
MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) {
setRequestMapper(messageMapper);
}

4
spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml
View File

@@ -91,8 +91,8 @@
<header name="foo" value="bar"/>
</method>
<method name="oneWayWithTimeouts" request-channel="otherRequestChannel"
request-timeout="#args[1]"
reply-timeout="#args[2]">
request-timeout="args[1]"
reply-timeout="args[2]">
</method>
</gateway>

12
spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java
View File

@@ -108,8 +108,8 @@ public class GatewayParserTests {
assertThat(meta.getReplyChannelName()).isEqualTo("foo");
meta = (GatewayMethodMetadata) methods.get("oneWayWithTimeouts");
assertThat(meta).isNotNull();
assertThat(meta.getRequestTimeout()).isEqualTo("#args[1]");
assertThat(meta.getReplyTimeout()).isEqualTo("#args[2]");
assertThat(meta.getRequestTimeout()).isEqualTo("args[1]");
assertThat(meta.getReplyTimeout()).isEqualTo("args[2]");
service.oneWayWithTimeouts("foo", 100L, 200L);
result = channel.receive(10000);
assertThat(result).isNotNull();
@@ -118,7 +118,7 @@ public class GatewayParserTests {
@Test
public void testSolicitResponse() {
PollableChannel channel = (PollableChannel) context.getBean("replyChannel");
channel.send(new GenericMessage<String>("foo"));
channel.send(new GenericMessage<>("foo"));
TestService service = (TestService) context.getBean("solicitResponse");
String result = service.solicitResponse();
assertThat(result).isEqualTo("foo");
@@ -161,7 +161,7 @@ public class GatewayParserTests {
}
@Test
public void testFactoryBeanObjectTypeWithServiceInterface() throws Exception {
public void testFactoryBeanObjectTypeWithServiceInterface() {
ConfigurableListableBeanFactory beanFactory = ((GenericApplicationContext) context).getBeanFactory();
Object attribute = beanFactory.getMergedBeanDefinition("&oneWay").getAttribute(
IntegrationConfigUtils.FACTORY_BEAN_OBJECT_TYPE);
@@ -169,7 +169,7 @@ public class GatewayParserTests {
}
@Test
public void testFactoryBeanObjectTypeWithNoServiceInterface() throws Exception {
public void testFactoryBeanObjectTypeWithNoServiceInterface() {
ConfigurableListableBeanFactory beanFactory = ((GenericApplicationContext) context).getBeanFactory();
Object attribute = beanFactory.getMergedBeanDefinition("&defaultConfig").getAttribute(
IntegrationConfigUtils.FACTORY_BEAN_OBJECT_TYPE);
@@ -177,7 +177,7 @@ public class GatewayParserTests {
}
@Test
public void testMonoGateway() throws Exception {
public void testMonoGateway() {
PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class);
MessageChannel replyChannel = context.getBean("replyChannel", MessageChannel.class);
this.startResponder(requestChannel, replyChannel);

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

@@ -810,7 +810,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow errorRecovererFlow() {
return IntegrationFlows.from(Function.class, "errorRecovererFunction")
return IntegrationFlows.from(Function.class, (gateway) -> gateway.beanName("errorRecovererFunction"))
.handle((GenericHandler<?>) (p, h) -> {
throw new RuntimeException("intentional");
}, e -> e.advice(retryAdvice()))
@@ -891,7 +891,8 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow globalErrorChannelResolutionFlow(@Qualifier("taskScheduler") TaskExecutor taskExecutor) {
return IntegrationFlows.from(Consumer.class, "globalErrorChannelResolutionFunction")
return IntegrationFlows.from(Consumer.class,
(gateway) -> gateway.beanName("globalErrorChannelResolutionFunction"))
.channel(c -> c.executor(taskExecutor))
.handle((GenericHandler<?>) (p, h) -> {
throw new RuntimeException("intentional");

59
spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java
View File

@@ -19,6 +19,9 @@ package org.springframework.integration.dsl.gateway;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import java.lang.reflect.Method;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +35,7 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.gateway.MethodArgsHolder;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -97,6 +101,27 @@ public class GatewayDslTests {
.withStackTraceContaining("intentional");
}
@Autowired
private Function<Object, Message<?>> functionGateay;
@Test
void testHeadersFromFunctionGateway() {
Message<?> message = this.functionGateay.apply("testPayload");
assertThat(message.getPayload()).isEqualTo("testPayload");
assertThat(message.getHeaders()).containsKeys("gatewayMethod", "gatewayArgs");
}
@Autowired
private RoutingGateway routingGateway;
@Test
void testRoutingGateway() {
String result = this.routingGateway.route1("test1");
assertThat(result).isEqualTo("route1");
result = this.routingGateway.route2("test2");
assertThat(result).isEqualTo("route2");
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@@ -134,6 +159,40 @@ public class GatewayDslTests {
})));
}
@Bean
public IntegrationFlow functionGateway() {
return IntegrationFlows.from(MessageFunction.class,
(gateway) -> gateway
.header("gatewayMethod", MethodArgsHolder::getMethod)
.header("gatewayArgs", MethodArgsHolder::getArgs))
.bridge()
.get();
}
@Bean
public IntegrationFlow routingGateway() {
return IntegrationFlows.from(RoutingGateway.class,
(gateway) -> gateway.header("gatewayMethod", MethodArgsHolder::getMethod))
.route(Message.class, (message) ->
message.getHeaders().get("gatewayMethod", Method.class).getName(),
(router) -> router
.subFlowMapping("route1", (subFlow) -> subFlow.transform((payload) -> "route1"))
.subFlowMapping("route2", (subFlow) -> subFlow.transform((payload) -> "route2")))
.get();
}
}
interface MessageFunction extends Function<Object, Message<?>> {
}
interface RoutingGateway {
String route1(Object payload);
String route2(Object payload);
}
}

9
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapperToMessageTests.java
View File

@@ -29,6 +29,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.mapping.MessageMappingException;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@@ -236,7 +237,7 @@ public class GatewayMethodInboundMessageMapperToMessageTests {
map.put(2, "Two");
GatewayMethodInboundMessageMapper mapper = new GatewayMethodInboundMessageMapper(method);
mapper.setBeanFactory(mock(BeanFactory.class));
mapper.setPayloadExpression("'hello'");
mapper.setPayloadExpression(new LiteralExpression("hello"));
Message<?> message = mapper.toMessage(new Object[] { map });
assertThat(message.getPayload()).isEqualTo("hello");
}
@@ -244,12 +245,12 @@ public class GatewayMethodInboundMessageMapperToMessageTests {
@Test
public void toMessageWithNonHeaderMapPayloadExpressionB() throws Exception {
Method method = TestService.class.getMethod("sendNonHeadersMap", Map.class);
Map<Integer, Object> map = new HashMap<Integer, Object>();
Map<Integer, Object> map = new HashMap<>();
map.put(1, "One");
map.put(2, "Two");
GatewayMethodInboundMessageMapper mapper = new GatewayMethodInboundMessageMapper(method);
mapper.setBeanFactory(mock(BeanFactory.class));
mapper.setPayloadExpression("#args[0]");
mapper.setPayloadExpression(new FunctionExpression<MethodArgsHolder>((methodArgs) -> methodArgs.getArgs()[0]));
Message<?> message = mapper.toMessage(new Object[] { map });
assertThat(message.getPayload()).isEqualTo(map);
}
@@ -277,7 +278,7 @@ public class GatewayMethodInboundMessageMapperToMessageTests {
mapB.put("2", "TWO");
GatewayMethodInboundMessageMapper mapper = new GatewayMethodInboundMessageMapper(method);
mapper.setBeanFactory(mock(BeanFactory.class));
mapper.setPayloadExpression("#args[0]");
mapper.setPayloadExpression(new FunctionExpression<MethodArgsHolder>((methodArgs) -> methodArgs.getArgs()[0]));
Message<?> message = mapper.toMessage(new Object[] { mapA, mapB });
assertThat(message.getPayload()).isEqualTo(mapA);
assertThat(message.getHeaders().get("1")).isEqualTo(mapB.get("1"));

7
src/reference/asciidoc/dsl.adoc
View File

@@ -1171,7 +1171,10 @@ Nevertheless, the `requestChannel` is ignored and overridden with that internal
Otherwise, creating such a configuration by using `IntegrationFlow` does not make sense.
By default a `GatewayProxyFactoryBean` gets a conventional bean name, such as `[FLOW_BEAN_NAME.gateway]`.
You can change that ID by using the `@MessagingGateway.name()` attribute or the overloaded `from(Class<?> serviceInterface, String beanName)` factory method.
You can change that ID by using the `@MessagingGateway.name()` attribute or the overloaded `IntegrationFlows.from(Class<?> serviceInterface, Consumer<GatewayProxySpec> endpointConfigurer)` factory method.
Also all the attributes from the `@MessagingGateway` annotation on the interface are applied to the target `GatewayProxyFactoryBean`.
When annotation configuration is not applicable, the `Consumer<GatewayProxySpec>` variant can be used for providing appropriate option for the target proxy.
This DSL method is available starting with version 5.2; the method `IntegrationFlows.from(Class<?> serviceInterface, String beanName)` is deprecated in favor of `GatewayProxySpec.beanName()` option.
With Java 8, you can even create an integration gateway with the `java.util.function` interfaces, as the following example shows:
@@ -1180,7 +1183,7 @@ With Java 8, you can even create an integration gateway with the `java.util.func
----
@Bean
public IntegrationFlow errorRecovererFlow() {
return IntegrationFlows.from(Function.class, "errorRecovererFunction")
return IntegrationFlows.from(Function.class, (gateway) -> gateway.beanName("errorRecovererFunction"))
.handle((GenericHandler<?>) (p, h) -> {
throw new RuntimeException("intentional");
}, e -> e.advice(retryAdvice()))

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

@@ -156,7 +156,9 @@ In the preceding example a different value is set for the 'RESPONSE_TYPE' header
The `<header/>` element supports `expression` as an alternative to `value`.
The SpEL expression is evaluated to determine the value of the header.
There is no `#root` object, but the following variables are available:
Starting with version 5.2, the `#root` object of the evaluation context is a `MethodArgsHolder` with `getMethod()` and `getArgs()` accessors.
These two expression evaluation context variables are deprecated since version 5.2:
* #args: An `Object[]` containing the method arguments
* #gatewayMethod: The object (derived from `java.reflect.Method`) that represents the method in the `service-interface` that was invoked.
@@ -164,8 +166,8 @@ A header containing this variable can be used later in the flow (for example, fo
For example, if you wish to route on the simple method name, you might add a header with the following expression: `#gatewayMethod.name`.
NOTE: The `java.reflect.Method` is not serializable.
A header with an expression of `#gatewayMethod` is lost if you later serialize the message.
Consequently, you may wish to use `#gatewayMethod.name` or `#gatewayMethod.toString()` in those cases.
A header with an expression of `method` is lost if you later serialize the message.
Consequently, you may wish to use `method.name` or `method.toString()` in those cases.
The `toString()` method provides a `String` representation of the method, including parameter and return types.
Since version 3.0, `<default-header/>` elements can be defined to add headers to all the messages produced by the gateway, regardless of the method invoked.

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -12,7 +12,7 @@ If you are interested in the changes and features that were introduced in earlie
If you are interested in more details, see the Issue Tracker tickets that were resolved as part of the 5.2 development process.
[[x5.2-package-clas]]
[[x5.2-package-class]]
=== Package and Class Changes
`Pausable` has been moved from `o.s.i.endpoint` to `o.s.i.core`.
@@ -86,6 +86,12 @@ See <<./aggregator.adoc#aggregator-api,Aggregator Programming Model>> for more i
All the `MessageHandlingException` s thrown in the framework, includes now a bean resource and source for back tracking a configuration part in case no end-user code involved.
See <<./error-handling.adoc#error-handling,Error Handling>> for more information.
For better end-user experience, Java DSL now provides a configurer variant for starting flow with a gateway interface.
See `IntegrationFlows.from(Class<?> serviceInterface, Consumer<GatewayProxySpec> endpointConfigurer)` JavaDocs for more information.
Also a `MethodArgsHolder` is now a root object for evaluation context for all the expressions in the `GatewayProxyFactoryBean`.
The `#args` and `#method` evaluation context variables are now deprecated.
See <<./gateway.adoc#gateway,Messaging Gateways>> for more information.
[[x5.2-amqp]]
==== AMQP Changes