diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java index e86c40feea..d1f5e37396 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java @@ -23,7 +23,6 @@ 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; @@ -161,7 +160,7 @@ public class GatewayProxySpec { * 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 + * Use {@code 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}. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java index 3f7e086d8c..6765e714c3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java @@ -111,8 +111,10 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper returnType = method.getReturnType(); - if (gateway.isReturnTypeMessage == null) { - gateway.isReturnTypeMessage = - Message.class.isAssignableFrom(returnType) || hasReturnMessageTypeOnFunction(method); + if (gateway.getReturnTypeMessage() == null) { + gateway.setReturnTypeMessage(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; + Object response; boolean hasPayloadExpression = findPayloadExpression(method); if (paramCount == 0 && !hasPayloadExpression) { - Long receiveTimeout = null; - if (gateway.getReceiveTimeoutExpression() != null) { - receiveTimeout = gateway.getReceiveTimeoutExpression().getValue(this.evaluationContext, Long.class); - } - if (shouldReply) { - if (shouldReturnMessage) { - if (receiveTimeout != null) { - return gateway.receiveMessage(receiveTimeout); - } - else { - return gateway.receiveMessage(); - } - } - if (receiveTimeout != null) { - response = gateway.receive(receiveTimeout); - } - else { - response = gateway.receive(); - } - } + response = receive(gateway, method, shouldReply, shouldReturnMessage); } else { response = sendOrSendAndReceive(invocation, gateway, shouldReturnMessage, shouldReply); @@ -557,6 +535,35 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint return hasPayloadExpression; } + @Nullable + private Object receive(MethodInvocationGateway gateway, Method method, boolean shouldReply, + boolean shouldReturnMessage) { + + Long receiveTimeout = null; + Expression receiveTimeoutExpression = gateway.getReceiveTimeoutExpression(); + if (receiveTimeoutExpression != null) { + receiveTimeout = receiveTimeoutExpression.getValue(this.evaluationContext, Long.class); + } + if (shouldReply) { + if (shouldReturnMessage) { + if (receiveTimeout != null) { + return gateway.receiveMessage(receiveTimeout); + } + else { + return gateway.receiveMessage(); + } + } + if (receiveTimeout != null) { + return gateway.receive(receiveTimeout); + } + else { + return gateway.receive(); + } + } + throw new IllegalArgumentException("The 'void' method without arguments '" + method + "' is not eligible for" + + " gateway invocation. Consider to use different signature or 'payloadExpression'."); + } + @Nullable private Object sendOrSendAndReceive(MethodInvocation invocation, MethodInvocationGateway gateway, boolean shouldReturnMessage, boolean shouldReply) { @@ -595,17 +602,18 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private MethodInvocationGateway createGatewayForMethod(Method method) { Gateway gatewayAnnotation = method.getAnnotation(Gateway.class); - String requestChannelName = null; - String replyChannelName = null; + GatewayMethodMetadata methodMetadata = null; + if (!CollectionUtils.isEmpty(this.methodMetadataMap)) { + methodMetadata = this.methodMetadataMap.get(method.getName()); + } + Map headerExpressions = new HashMap<>(); Expression requestTimeout = this.defaultRequestTimeout; Expression replyTimeout = this.defaultReplyTimeout; - Expression payloadExpression = this.globalMethodMetadata != null - ? this.globalMethodMetadata.getPayloadExpression() - : null; - Map headerExpressions = new HashMap<>(); + Expression payloadExpression = + extractPayloadExpressionFromAnnotationOrMetadata(gatewayAnnotation, methodMetadata); + String requestChannelName = extractRequestChannelFromAnnotationOrMetadata(gatewayAnnotation, methodMetadata); + String replyChannelName = extractReplyChannelFromAnnotationOrMetadata(gatewayAnnotation, methodMetadata); if (gatewayAnnotation != null) { - requestChannelName = gatewayAnnotation.requestChannel(); - replyChannelName = gatewayAnnotation.replyChannel(); /* * INT-2636 Unspecified annotation attributes should not * override the default values supplied by explicit configuration. @@ -625,62 +633,74 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (StringUtils.hasText(gatewayAnnotation.replyTimeoutExpression())) { replyTimeout = ExpressionUtils.longExpression(gatewayAnnotation.replyTimeoutExpression()); } - if (payloadExpression == null && StringUtils.hasText(gatewayAnnotation.payloadExpression())) { - payloadExpression = PARSER.parseExpression(gatewayAnnotation.payloadExpression()); - } annotationHeaders(gatewayAnnotation, headerExpressions); } - else if (this.methodMetadataMap != null && this.methodMetadataMap.size() > 0) { - GatewayMethodMetadata methodMetadata = this.methodMetadataMap.get(method.getName()); - if (methodMetadata != null) { - if (methodMetadata.getPayloadExpression() != null) { - payloadExpression = methodMetadata.getPayloadExpression(); - } - if (!CollectionUtils.isEmpty(methodMetadata.getHeaderExpressions())) { - headerExpressions.putAll(methodMetadata.getHeaderExpressions()); - } - requestChannelName = methodMetadata.getRequestChannelName(); - replyChannelName = methodMetadata.getReplyChannelName(); - String reqTimeout = methodMetadata.getRequestTimeout(); - if (StringUtils.hasText(reqTimeout)) { - requestTimeout = ExpressionUtils.longExpression(reqTimeout); - } - String repTimeout = methodMetadata.getReplyTimeout(); - if (StringUtils.hasText(repTimeout)) { - replyTimeout = ExpressionUtils.longExpression(repTimeout); - } + else if (methodMetadata != null) { + if (!CollectionUtils.isEmpty(methodMetadata.getHeaderExpressions())) { + headerExpressions.putAll(methodMetadata.getHeaderExpressions()); + } + String reqTimeout = methodMetadata.getRequestTimeout(); + if (StringUtils.hasText(reqTimeout)) { + requestTimeout = ExpressionUtils.longExpression(reqTimeout); + } + String repTimeout = methodMetadata.getReplyTimeout(); + if (StringUtils.hasText(repTimeout)) { + replyTimeout = ExpressionUtils.longExpression(repTimeout); } } - Map headers = headers(method, headerExpressions); - GatewayMethodInboundMessageMapper messageMapper = new GatewayMethodInboundMessageMapper(method, - headerExpressions, - this.globalMethodMetadata != null ? this.globalMethodMetadata.getHeaderExpressions() : null, - headers, this.argsMapper, getMessageBuilderFactory()); + return doCreateMethodInvocationGateway(method, payloadExpression, headerExpressions, + requestChannelName, replyChannelName, requestTimeout, replyTimeout); + } - MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper); + @Nullable + private Expression extractPayloadExpressionFromAnnotationOrMetadata(@Nullable Gateway gatewayAnnotation, + @Nullable GatewayMethodMetadata methodMetadata) { - JavaUtils.INSTANCE - .acceptIfNotNull(payloadExpression, messageMapper::setPayloadExpression) - .acceptIfNotNull(getTaskScheduler(), gateway::setTaskScheduler); - gateway.setBeanName(getComponentName()); + Expression payloadExpression = + this.globalMethodMetadata != null + ? this.globalMethodMetadata.getPayloadExpression() + : null; - setChannel(this.errorChannel, gateway::setErrorChannel, this.errorChannelName, gateway::setErrorChannelName); - setChannel(requestChannelName, this.defaultRequestChannelName, gateway::setRequestChannelName, - this.defaultRequestChannel, gateway::setRequestChannel); - setChannel(replyChannelName, this.defaultReplyChannelName, gateway::setReplyChannelName, - this.defaultReplyChannel, gateway::setReplyChannel); - - timeouts(requestTimeout, replyTimeout, messageMapper, gateway); - BeanFactory beanFactory = getBeanFactory(); - if (beanFactory != null) { - gateway.setBeanFactory(beanFactory); - messageMapper.setBeanFactory(beanFactory); + if (gatewayAnnotation != null) { + if (payloadExpression == null && StringUtils.hasText(gatewayAnnotation.payloadExpression())) { + payloadExpression = PARSER.parseExpression(gatewayAnnotation.payloadExpression()); + } } - gateway.setShouldTrack(this.shouldTrack); - gateway.afterPropertiesSet(); - return gateway; + else if (methodMetadata != null) { + if (methodMetadata.getPayloadExpression() != null) { + payloadExpression = methodMetadata.getPayloadExpression(); + } + } + + return payloadExpression; + } + + @Nullable + private String extractRequestChannelFromAnnotationOrMetadata(@Nullable Gateway gatewayAnnotation, + @Nullable GatewayMethodMetadata methodMetadata) { + + if (gatewayAnnotation != null) { + return gatewayAnnotation.requestChannel(); + } + else if (methodMetadata != null) { + return methodMetadata.getRequestChannelName(); + } + return null; + } + + @Nullable + private String extractReplyChannelFromAnnotationOrMetadata(@Nullable Gateway gatewayAnnotation, + @Nullable GatewayMethodMetadata methodMetadata) { + + if (gatewayAnnotation != null) { + return gatewayAnnotation.replyChannel(); + } + else if (methodMetadata != null) { + return methodMetadata.getReplyChannelName(); + } + return null; } private void annotationHeaders(Gateway gatewayAnnotation, Map headerExpressions) { @@ -702,6 +722,41 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } + private MethodInvocationGateway doCreateMethodInvocationGateway(Method method, + @Nullable Expression payloadExpression, Map headerExpressions, + @Nullable String requestChannelName, @Nullable String replyChannelName, + Expression requestTimeout, Expression replyTimeout) { + + GatewayMethodInboundMessageMapper messageMapper = createGatewayMessageMapper(method, headerExpressions); + MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper); + + JavaUtils.INSTANCE + .acceptIfNotNull(payloadExpression, messageMapper::setPayloadExpression) + .acceptIfNotNull(getTaskScheduler(), gateway::setTaskScheduler); + + channels(requestChannelName, replyChannelName, gateway); + + timeouts(requestTimeout, replyTimeout, messageMapper, gateway); + + gateway.setBeanName(getComponentName()); + gateway.setBeanFactory(getBeanFactory()); + gateway.setShouldTrack(this.shouldTrack); + gateway.afterPropertiesSet(); + + return gateway; + } + + private GatewayMethodInboundMessageMapper createGatewayMessageMapper(Method method, Map headerExpressions) { + + Map headers = headers(method, headerExpressions); + + return new GatewayMethodInboundMessageMapper(method, + headerExpressions, + this.globalMethodMetadata != null ? this.globalMethodMetadata.getHeaderExpressions() : null, + headers, this.argsMapper, getMessageBuilderFactory()); + } + @Nullable private Map headers(Method method, Map headerExpressions) { Map headers = null; @@ -744,7 +799,17 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } - private void timeouts(Expression requestTimeout, Expression replyTimeout, + private void channels(@Nullable String requestChannelName, @Nullable String replyChannelName, + MethodInvocationGateway gateway) { + + setChannel(this.errorChannel, gateway::setErrorChannel, this.errorChannelName, gateway::setErrorChannelName); + setChannel(requestChannelName, this.defaultRequestChannelName, gateway::setRequestChannelName, + this.defaultRequestChannel, gateway::setRequestChannel); + setChannel(replyChannelName, this.defaultReplyChannelName, gateway::setReplyChannelName, + this.defaultReplyChannel, gateway::setReplyChannel); + } + + private void timeouts(@Nullable Expression requestTimeout, @Nullable Expression replyTimeout, GatewayMethodInboundMessageMapper messageMapper, MethodInvocationGateway gateway) { if (requestTimeout == null) { gateway.setRequestTimeout(-1); @@ -869,12 +934,13 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private Expression receiveTimeoutExpression; - volatile Boolean isReturnTypeMessage; + private volatile Boolean isReturnTypeMessage; MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) { setRequestMapper(messageMapper); } + @Nullable Expression getReceiveTimeoutExpression() { return this.receiveTimeoutExpression; } @@ -883,6 +949,15 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint this.receiveTimeoutExpression = receiveTimeoutExpression; } + @Nullable + Boolean getReturnTypeMessage() { + return this.isReturnTypeMessage; + } + + void setReturnTypeMessage(Boolean returnTypeMessage) { + this.isReturnTypeMessage = returnTypeMessage; + } + } private final class Invoker implements Supplier { @@ -902,7 +977,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint throw e; } catch (Throwable t) { //NOSONAR - if (t instanceof RuntimeException) { + if (t instanceof RuntimeException) { //NOSONAR throw (RuntimeException) t; } throw new MessagingException("Asynchronous gateway invocation failed", t); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 79174e273d..624e5ff589 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -23,6 +23,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.reactivestreams.Subscriber; import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.core.AttributeAccessor; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; @@ -349,6 +350,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint if (this.requestMapper instanceof DefaultRequestMapper) { ((DefaultRequestMapper) this.requestMapper).setMessageBuilderFactory(messageBuilderFactory); } + if (this.requestMapper instanceof BeanFactoryAware) { + ((BeanFactoryAware) this.requestMapper).setBeanFactory(beanFactory); + } this.messageConverter.setBeanFactory(beanFactory); } this.initialized = true;