From f58106ec3cda2b1b6c828d3892ba683989d07eee Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 23 May 2017 21:28:52 -0400 Subject: [PATCH] INT-4033: Support SpEL in Gateway Timeouts JIRA: https://jira.spring.io/browse/INT-4033 Provide a mechanism to support dynamic timeouts when invoking gateway methods. Polishing - PR Comments * Polishing `what's new` * Add `STOMP` to the `endpoint-summary.adoc` * Polishing code style a bit for the changes in this fix --- .../integration/annotation/Gateway.java | 30 +++- .../annotation/MessagingGateway.java | 19 ++- .../config/MessagingGatewayRegistrar.java | 6 +- .../integration/core/MessagingTemplate.java | 16 +- .../expression/ExpressionUtils.java | 38 +++++ .../expression/ValueExpression.java | 6 + .../GatewayMethodInboundMessageMapper.java | 71 +++++---- .../gateway/GatewayProxyFactoryBean.java | 137 +++++++++++++++--- .../gateway/MessagingGatewaySupport.java | 16 ++ .../config/spring-integration-5.0.xsd | 14 +- .../integration/config/ChainParserTests.java | 7 +- .../config/xml/GatewayParserTests-context.xml | 6 + .../config/xml/GatewayParserTests.java | 11 ++ .../gateway/GatewayInterfaceTests-context.xml | 1 - .../gateway/GatewayInterfaceTests.java | 24 +-- .../gateway/GatewayXmlAndAnnotationTests.java | 7 +- .../integration/gateway/TestService.java | 3 + src/reference/asciidoc/endpoint-summary.adoc | 25 +++- src/reference/asciidoc/gateway.adoc | 31 ++++ src/reference/asciidoc/whats-new.adoc | 52 ++++--- 20 files changed, 421 insertions(+), 99 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Gateway.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Gateway.java index 4aa5436ee7..2f168ac892 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Gateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Gateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 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. @@ -57,7 +57,7 @@ import java.lang.annotation.Target; * * @see MessagingGateway */ -@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) +@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE }) @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented @@ -80,19 +80,41 @@ public @interface Gateway { /** * Specify the timeout (ms) when sending to the request channel - only applies if the * send might block (such as a bounded {@code QueueChannel} that is currently full. - * Overrides the encompassing gatewsy's default request timeout. + * Overrides the encompassing gateway's default request timeout. * @return the timeout. + * @see #requestTimeoutExpression() */ long requestTimeout() default Long.MIN_VALUE; + /** + * Specify a SpEL Expression to determine the timeout (ms) when sending to the request + * channel - only applies if the send might block (such as a bounded + * {@code QueueChannel} that is currently full. Overrides the encompassing gateway's + * default request timeout. Overrides {@link #requestTimeout()}. + * @return the timeout. + * @since 5.0 + */ + String requestTimeoutExpression() default ""; + /** * Specify the time (ms) that the thread sending the request will wait for a reply. * The timer starts when the thread returns to the gateway, not when the request * message is sent. Overrides the encompassing gateway's default reply timeout. * @return the timeout. + * @see #replyTimeoutExpression() */ long replyTimeout() default Long.MIN_VALUE; + /** + * Specify a SpEL Expression to determine the the time (ms) that the thread sending + * the request will wait for a reply. The timer starts when the thread returns to the + * gateway, not when the request message is sent. Overrides the encompassing gateway's + * default reply timeout. Overrides {@link #replyTimeout()}. + * @return the timeout. + * @since 5.0 + */ + String replyTimeoutExpression() default ""; + /** * Specify a SpEL expression to determine the payload of the request message. * @return the expression. @@ -103,6 +125,6 @@ public @interface Gateway { * Specify additional headers that will be added to the request message. * @return the headers. */ - GatewayHeader[] headers() default {}; + GatewayHeader[] headers() default { }; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java index 6f9280b611..db18b685f5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -35,6 +35,8 @@ import java.lang.annotation.Target; * ignores interfaces. * * @author Artem Bilan + * @author Gary Russell + * * @since 4.0 * * @see IntegrationComponentScan @@ -76,17 +78,20 @@ public @interface MessagingGateway { String errorChannel() default ""; /** - * 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}. + * 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; it can be a simple long value or a SpEL expression; array variable + * #args is available. * @return the suggested timeout in milliseconds, if any */ String defaultRequestTimeout() default "-9223372036854775808"; /** * 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. + * before returning. By default it will wait indefinitely. {@code null} is returned if + * the gateway times out. Value is specified in milliseconds; it can be a simple long + * value or a SpEL expression; array variable #args is available. * @return the suggested timeout in milliseconds, if any */ String defaultReplyTimeout() default "-9223372036854775808"; @@ -116,7 +121,7 @@ public @interface MessagingGateway { * all methods on the service-interface (unless overridden by a specific method). * @return the suggested payload expression, if any */ - GatewayHeader[] defaultHeaders() default {}; + GatewayHeader[] defaultHeaders() default { }; /** * An {@link org.springframework.integration.gateway.MethodArgsMessageMapper} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java index 97c5500696..89336ca641 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java @@ -145,8 +145,10 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar gatewayProxyBuilder.addPropertyReference("mapper", mapper); } - gatewayProxyBuilder.addPropertyValue("defaultRequestTimeout", gatewayAttributes.get("defaultRequestTimeout")); - gatewayProxyBuilder.addPropertyValue("defaultReplyTimeout", gatewayAttributes.get("defaultReplyTimeout")); + gatewayProxyBuilder.addPropertyValue("defaultRequestTimeoutExpressionString", + gatewayAttributes.get("defaultRequestTimeout")); + gatewayProxyBuilder.addPropertyValue("defaultReplyTimeoutExpressionString", + gatewayAttributes.get("defaultReplyTimeout")); gatewayProxyBuilder.addPropertyValue("methodMetadataMap", gatewayAttributes.get("methods")); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 68bd51ab4d..d8888a15a5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -97,4 +97,18 @@ public class MessagingTemplate extends GenericMessagingTemplate { return super.sendAndReceive(destination, requestMessage); } + public Object receiveAndConvert(MessageChannel destination, long timeout) { + Message message = doReceive(destination, timeout); + if (message != null) { + return doConvert(message, null); + } + else { + return null; + } + } + + public Message receive(MessageChannel destination, long timeout) { + return doReceive(destination, timeout); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java index 91bd877e58..8012c54708 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java @@ -27,6 +27,8 @@ import org.springframework.context.expression.MapAccessor; import org.springframework.core.convert.ConversionService; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; +import org.springframework.expression.ExpressionParser; +import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.expression.spel.support.StandardTypeConverter; import org.springframework.integration.context.IntegrationContextUtils; @@ -44,6 +46,8 @@ import org.springframework.util.Assert; */ public final class ExpressionUtils { + private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser(); + private static final Log logger = LogFactory.getLog(ExpressionUtils.class); private ExpressionUtils() { @@ -141,4 +145,38 @@ public final class ExpressionUtils { return file; } + /** + * Return a {@link ValueExpression} for a simple literal, otherwise + * a {@link org.springframework.expression.spel.standard.SpelExpression}. + * @param expression the expression string. + * @return the expression. + * @since 5.0 + */ + public static Expression intExpression(String expression) { + try { + return new ValueExpression<>(Integer.parseInt(expression)); + } + catch (NumberFormatException e) { + // empty + } + return EXPRESSION_PARSER.parseExpression(expression); + } + + /** + * Return a {@link ValueExpression} for a simple literal, otherwise + * a {@link org.springframework.expression.spel.standard.SpelExpression}. + * @param expression the expression string. + * @return the expression. + * @since 5.0 + */ + public static Expression longExpression(String expression) { + try { + return new ValueExpression<>(Long.parseLong(expression)); + } + catch (NumberFormatException e) { + // empty + } + return EXPRESSION_PARSER.parseExpression(expression); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/expression/ValueExpression.java b/spring-integration-core/src/main/java/org/springframework/integration/expression/ValueExpression.java index 0e7b272bc3..97195a7843 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/expression/ValueExpression.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/expression/ValueExpression.java @@ -30,6 +30,7 @@ import org.springframework.util.Assert; * @param - The expected value type. * * @author Artem Bilan + * @author Gary Russell * @since 4.0 */ public class ValueExpression implements Expression { @@ -169,4 +170,9 @@ public class ValueExpression implements Expression { return this.value.toString(); } + @Override + public String toString() { + return "ValueExpression [value=" + this.value + "]"; + } + } 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 4b398b0de5..8d13066eaf 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 @@ -47,6 +47,7 @@ import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.util.MessagingAnnotationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.core.GenericMessagingTemplate; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Headers; import org.springframework.messaging.handler.annotation.Payload; @@ -77,11 +78,12 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * * @since 2.0 */ class GatewayMethodInboundMessageMapper implements InboundMessageMapper, BeanFactoryAware { - private final Log logger = LogFactory.getLog(this.getClass()); + private final static Log logger = LogFactory.getLog(GatewayMethodInboundMessageMapper.class); private static final SpelExpressionParser PARSER = new SpelExpressionParser(); @@ -97,6 +99,8 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper parameterPayloadExpressions = new HashMap(); @@ -105,7 +109,9 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper evaluateHeaders(EvaluationContext methodInvocationEvaluationContext, Map headerExpressions) { - Map evaluatedHeaders = new HashMap(); + Map evaluatedHeaders = new HashMap<>(); for (Map.Entry entry : headerExpressions.entrySet()) { Object value = entry.getValue().getValue(methodInvocationEvaluationContext); - if (value != null) { - evaluatedHeaders.put(entry.getKey(), value); - } + evaluatedHeaders.put(entry.getKey(), value); } return evaluatedHeaders; } @@ -205,11 +215,8 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper entry : argumentValue.entrySet()) { Object key = entry.getKey(); if (!(key instanceof String)) { - if (this.logger.isWarnEnabled()) { - this.logger.warn("Invalid header name [" + key + + if (logger.isWarnEnabled()) { + logger.warn("Invalid header name [" + key + "], name type must be String. Skipping mapping of this header to MessageHeaders."); } } @@ -232,7 +239,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper getMethodParameterList(Method method) { - List parameterList = new LinkedList(); + List parameterList = new LinkedList<>(); ParameterNameDiscoverer parameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer(); int parameterCount = method.getParameterTypes().length; for (int i = 0; i < parameterCount; i++) { @@ -335,14 +342,26 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper) argumentValue, headers); } else if (GatewayMethodInboundMessageMapper.this.payloadExpression == null) { - GatewayMethodInboundMessageMapper.this.throwExceptionForMultipleMessageOrPayloadParameters(methodParameter); + GatewayMethodInboundMessageMapper.this + .throwExceptionForMultipleMessageOrPayloadParameters(methodParameter); } } - Assert.isTrue(messageOrPayload != null, - "unable to determine a Message or payload parameter on method [" + GatewayMethodInboundMessageMapper.this.method + "]"); - AbstractIntegrationMessageBuilder builder = (messageOrPayload instanceof Message) - ? GatewayMethodInboundMessageMapper.this.messageBuilderFactory.fromMessage((Message) messageOrPayload) - : GatewayMethodInboundMessageMapper.this.messageBuilderFactory.withPayload(messageOrPayload); + Assert.isTrue(messageOrPayload != null, "unable to determine a Message or payload parameter on method [" + + GatewayMethodInboundMessageMapper.this.method + "]"); + if (GatewayMethodInboundMessageMapper.this.sendTimeoutExpression != null) { + headers.computeIfAbsent(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER, + v -> GatewayMethodInboundMessageMapper.this.sendTimeoutExpression + .getValue(methodInvocationEvaluationContext, Long.class)); + } + if (GatewayMethodInboundMessageMapper.this.replyTimeoutExpression != null) { + headers.computeIfAbsent(GenericMessagingTemplate.DEFAULT_RECEIVE_TIMEOUT_HEADER, + v -> GatewayMethodInboundMessageMapper.this.replyTimeoutExpression + .getValue(methodInvocationEvaluationContext, Long.class)); + } + AbstractIntegrationMessageBuilder builder = + (messageOrPayload instanceof Message) + ? GatewayMethodInboundMessageMapper.this.messageBuilderFactory.fromMessage((Message) messageOrPayload) + : GatewayMethodInboundMessageMapper.this.messageBuilderFactory.withPayload(messageOrPayload); builder.copyHeadersIfAbsent(headers); // Explicit headers in XML override any @Header annotations... if (!CollectionUtils.isEmpty(GatewayMethodInboundMessageMapper.this.headerExpressions)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index c23d630e76..39338463a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -45,11 +45,15 @@ import org.springframework.core.task.AsyncListenableTaskExecutor; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; 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.support.StandardEvaluationContext; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.support.management.TrackableComponent; import org.springframework.integration.support.utils.IntegrationUtils; @@ -99,9 +103,9 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private volatile String errorChannelName; - private volatile Long defaultRequestTimeout; + private volatile Expression defaultRequestTimeout; - private volatile Long defaultReplyTimeout; + private volatile Expression defaultReplyTimeout; private volatile DestinationResolver channelResolver; @@ -131,6 +135,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private volatile MethodArgsMessageMapper argsMapper; + private EvaluationContext evaluationContext = new StandardEvaluationContext(); + /** * Create a Factory whose service interface type can be configured by setter injection. * If none is set, it will fall back to the default service interface type, @@ -223,25 +229,77 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } /** - * Set the default timeout value for sending request messages. If not - * explicitly configured with an annotation, this value will be used. + * Set the default timeout value for sending request messages. If not explicitly + * configured with an annotation, or on a method element, this value will be used. * * @param defaultRequestTimeout the timeout value in milliseconds */ public void setDefaultRequestTimeout(Long defaultRequestTimeout) { + this.defaultRequestTimeout = new ValueExpression<>(defaultRequestTimeout); + } + + /** + * Set an expression to be evaluated to determine the default timeout value for + * sending request messages. If not explicitly configured with an annotation, or on a + * method element, this value will be used. + * + * @param defaultRequestTimeout the timeout value in milliseconds + * @since 5.0 + */ + public void setDefaultRequestTimeoutExpression(Expression defaultRequestTimeout) { this.defaultRequestTimeout = defaultRequestTimeout; } /** - * Set the default timeout value for receiving reply messages. If not - * explicitly configured with an annotation, this value will be used. + * Set an expression to be evaluated to determine the default timeout value for + * sending request messages. If not explicitly configured with an annotation, or on a + * method element, this value will be used. + * + * @param defaultRequestTimeout the timeout value in milliseconds + * @since 5.0 + */ + public void setDefaultRequestTimeoutExpressionString(String defaultRequestTimeout) { + if (StringUtils.hasText(defaultRequestTimeout)) { + this.defaultRequestTimeout = ExpressionUtils.longExpression(defaultRequestTimeout); + } + } + + /** + * Set the default timeout value for receiving reply messages. If not explicitly + * configured with an annotation, or on a method element, this value will be used. * * @param defaultReplyTimeout the timeout value in milliseconds */ public void setDefaultReplyTimeout(Long defaultReplyTimeout) { + this.defaultReplyTimeout = new ValueExpression<>(defaultReplyTimeout); + } + + /** + * Set an expression to be evaluated to determine the default timeout value for + * receiving reply messages. If not explicitly configured with an annotation, or on a + * method element, this value will be used. + * + * @param defaultReplyTimeout the timeout value in milliseconds + * @since 5.0 + */ + public void setDefaultReplyTimeoutExpression(Expression defaultReplyTimeout) { this.defaultReplyTimeout = defaultReplyTimeout; } + /** + * Set an expression to be evaluated to determine the default timeout value for + * receiving reply messages. If not explicitly configured with an annotation, or on a + * method element, this value will be used. + * + * @param defaultReplyTimeout the timeout value in milliseconds + * @since 5.0 + */ + public void setDefaultReplyTimeoutExpressionString(String defaultReplyTimeout) { + if (StringUtils.hasText(defaultReplyTimeout)) { + this.defaultReplyTimeout = ExpressionUtils.longExpression(defaultReplyTimeout); + } + } + @Override public void setShouldTrack(boolean shouldTrack) { this.shouldTrack = shouldTrack; @@ -336,7 +394,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint this.asyncSubmitListenableType = submitType.getClass(); } } - + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); this.initialized = true; } } @@ -428,11 +486,25 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint hasPayloadExpression = (metadata != null) && StringUtils.hasText(metadata.getPayloadExpression()); } if (paramCount == 0 && !hasPayloadExpression) { + Long receiveTimeout = null; + if (gateway.getReceiveTimeoutExpression() != null) { + receiveTimeout = gateway.getReceiveTimeoutExpression().getValue(this.evaluationContext, Long.class); + } if (shouldReply) { if (shouldReturnMessage) { - return gateway.receiveMessage(); + if (receiveTimeout != null) { + return gateway.receiveMessage(receiveTimeout); + } + else { + return gateway.receiveMessage(); + } + } + if (receiveTimeout != null) { + response = gateway.receive(receiveTimeout); + } + else { + response = gateway.receive(); } - response = gateway.receive(); } } else { @@ -472,8 +544,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint Gateway gatewayAnnotation = method.getAnnotation(Gateway.class); String requestChannelName = null; String replyChannelName = null; - Long requestTimeout = this.defaultRequestTimeout; - Long replyTimeout = this.defaultReplyTimeout; + Expression requestTimeout = this.defaultRequestTimeout; + Expression replyTimeout = this.defaultReplyTimeout; String payloadExpression = this.globalMethodMetadata != null ? this.globalMethodMetadata.getPayloadExpression() : null; @@ -489,10 +561,16 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint * no longer work as expected; they will need to use, say, -1 instead. */ if (requestTimeout == null || gatewayAnnotation.requestTimeout() != Long.MIN_VALUE) { - requestTimeout = gatewayAnnotation.requestTimeout(); + requestTimeout = new ValueExpression<>(gatewayAnnotation.requestTimeout()); + } + if (StringUtils.hasText(gatewayAnnotation.requestTimeoutExpression())) { + requestTimeout = ExpressionUtils.longExpression(gatewayAnnotation.requestTimeoutExpression()); } if (replyTimeout == null || gatewayAnnotation.replyTimeout() != Long.MIN_VALUE) { - replyTimeout = gatewayAnnotation.replyTimeout(); + replyTimeout = new ValueExpression<>(gatewayAnnotation.replyTimeout()); + } + if (StringUtils.hasText(gatewayAnnotation.replyTimeoutExpression())) { + replyTimeout = ExpressionUtils.longExpression(gatewayAnnotation.replyTimeoutExpression()); } if (payloadExpression == null || StringUtils.hasText(gatewayAnnotation.payloadExpression())) { payloadExpression = gatewayAnnotation.payloadExpression(); @@ -529,11 +607,11 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint replyChannelName = methodMetadata.getReplyChannelName(); String reqTimeout = methodMetadata.getRequestTimeout(); if (StringUtils.hasText(reqTimeout)) { - requestTimeout = this.convert(reqTimeout, Long.class); + requestTimeout = ExpressionUtils.longExpression(reqTimeout); } String repTimeout = methodMetadata.getReplyTimeout(); if (StringUtils.hasText(repTimeout)) { - replyTimeout = this.convert(repTimeout, Long.class); + replyTimeout = ExpressionUtils.longExpression(repTimeout); } } } @@ -551,7 +629,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (StringUtils.hasText(payloadExpression)) { messageMapper.setPayloadExpression(payloadExpression); } - messageMapper.setBeanFactory(this.getBeanFactory()); + messageMapper.setBeanFactory(getBeanFactory()); MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper); if (this.errorChannel != null) { @@ -589,18 +667,27 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (requestTimeout == null) { gateway.setRequestTimeout(-1); } + else if (requestTimeout instanceof ValueExpression) { + gateway.setRequestTimeout(requestTimeout.getValue(Long.class)); + } else { - gateway.setRequestTimeout(requestTimeout); + messageMapper.setSendTimeoutExpression(requestTimeout); } if (replyTimeout == null) { gateway.setReplyTimeout(-1); } + else if (replyTimeout instanceof ValueExpression) { + gateway.setReplyTimeout(replyTimeout.getValue(Long.class)); + } else { - gateway.setReplyTimeout(replyTimeout); + messageMapper.setReplyTimeoutExpression(replyTimeout); } if (this.getBeanFactory() != null) { gateway.setBeanFactory(this.getBeanFactory()); } + if (replyTimeout != null) { + gateway.setReceiveTimeoutExpression(replyTimeout); + } gateway.setShouldTrack(this.shouldTrack); gateway.afterPropertiesSet(); return gateway; @@ -641,7 +728,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private static boolean hasReturnParameterizedWithMessage(Method method, boolean runningOnCallerThread) { if (!runningOnCallerThread && (Future.class.isAssignableFrom(method.getReturnType()) - || Mono.class.isAssignableFrom(method.getReturnType()))) { + || Mono.class.isAssignableFrom(method.getReturnType()))) { Type returnType = method.getGenericReturnType(); if (returnType instanceof ParameterizedType) { Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments(); @@ -662,8 +749,18 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private static final class MethodInvocationGateway extends MessagingGatewaySupport { + Expression receiveTimeoutExpression; + MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) { - this.setRequestMapper(messageMapper); + setRequestMapper(messageMapper); + } + + Expression getReceiveTimeoutExpression() { + return this.receiveTimeoutExpression; + } + + void setReceiveTimeoutExpression(Expression receiveTimeoutExpression) { + this.receiveTimeoutExpression = receiveTimeoutExpression; } } 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 7d4ca97a67..01a5d2e2a0 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 @@ -418,6 +418,22 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.messagingTemplate.receive(replyChannel); } + protected Object receive(long timeout) { + this.initializeIfNecessary(); + MessageChannel replyChannel = getReplyChannel(); + Assert.state(replyChannel != null && (replyChannel instanceof PollableChannel), + "receive is not supported, because no pollable reply channel has been configured"); + return this.messagingTemplate.receiveAndConvert(replyChannel, timeout); + } + + protected Message receiveMessage(long timeout) { + initializeIfNecessary(); + MessageChannel replyChannel = getReplyChannel(); + Assert.state(replyChannel instanceof PollableChannel, + "receive is not supported, because no pollable reply channel has been configured"); + return this.messagingTemplate.receive(replyChannel, timeout); + } + protected Object sendAndReceive(Object object) { return this.doSendAndReceive(object, true); } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd index 87459745fe..007fef328d 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd @@ -735,7 +735,8 @@ Provides the amount of time dispatcher would wait to send a 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 Queue channel.  - Value is specified in milliseconds. + Value is specified in milliseconds; it can be a simple long value or a SpEL + expression; array variable #args is available. ]]> @@ -747,7 +748,9 @@ Allows you to specify how long this gateway will wait for the reply message before returning. By default it will wait indefinitely. 'null' is returned if the gateway times out. - Value is specified in milliseconds. + Value is specified in milliseconds; it can be a simple long value or a SpEL + expression; array variable #args is available. + Also used for receive-only operations as the receive timeout. ]]> @@ -850,7 +853,8 @@ Provides the amount of time dispatcher would wait to send a 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 Queue channel.  - Value is specified in milliseconds. + Value is specified in milliseconds; it can be a simple long value or a SpEL + expression; array variable '#args' is available. ]]> @@ -871,7 +875,9 @@ Specifies how long this gateway will wait for the reply message before returning. By default it will wait indefinitely. 'null' is returned if the gateway times out. - Value is specified in milliseconds. + Value is specified in milliseconds; it can be a simple long value or a SpEL + expression; array variable '#args' is available. + Also used for receive-only operations as the receive timeout. ]]> diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java index 0d2536d660..428bce035b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java @@ -46,6 +46,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.channel.QueueChannel; @@ -391,8 +392,10 @@ public class ChainParserTests { GatewayProxyFactoryBean.class); assertEquals("strings", TestUtils.getPropertyValue(gatewayProxyFactoryBean, "defaultRequestChannelName")); assertEquals("numbers", TestUtils.getPropertyValue(gatewayProxyFactoryBean, "defaultReplyChannelName")); - assertEquals(new Long(1000), TestUtils.getPropertyValue(gatewayProxyFactoryBean, "defaultRequestTimeout", Long.class)); - assertEquals(new Long(100), TestUtils.getPropertyValue(gatewayProxyFactoryBean, "defaultReplyTimeout", Long.class)); + assertEquals(new Long(1000), TestUtils + .getPropertyValue(gatewayProxyFactoryBean, "defaultRequestTimeout", Expression.class).getValue()); + assertEquals(new Long(100), TestUtils + .getPropertyValue(gatewayProxyFactoryBean, "defaultReplyTimeout", Expression.class).getValue()); assertTrue(this.beanFactory.containsBean("subComponentsIdSupport1$child.objectToStringTransformerWithinChain.handler")); assertTrue(this.beanFactory.containsBean("subComponentsIdSupport1$child.objectToMapTransformerWithinChain.handler")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml index 0189e7a60f..043d6cf6af 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml @@ -79,6 +79,8 @@ service-interface="org.springframework.integration.gateway.TestService" default-request-channel="requestChannel" default-reply-channel="replyChannel" + default-request-timeout="1000" + default-reply-timeout="2000" async-executor="testExecutor">
+ + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index 477905431e..df77c05316 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -45,6 +45,7 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.expression.Expression; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.gateway.GatewayMethodMetadata; @@ -95,16 +96,26 @@ public class GatewayParserTests { service.oneWay("foo"); PollableChannel channel = (PollableChannel) context.getBean("otherRequestChannel"); Message result = channel.receive(10000); + assertNotNull(result); assertEquals("fiz", result.getPayload()); assertEquals("bar", result.getHeaders().get("foo")); assertEquals("qux", result.getHeaders().get("baz")); GatewayProxyFactoryBean fb = context.getBean("&methodOverride", GatewayProxyFactoryBean.class); + assertEquals(1000L, TestUtils.getPropertyValue(fb, "defaultRequestTimeout", Expression.class).getValue()); + assertEquals(2000L, TestUtils.getPropertyValue(fb, "defaultReplyTimeout", Expression.class).getValue()); Map methods = TestUtils.getPropertyValue(fb, "methodMetadataMap", Map.class); GatewayMethodMetadata meta = (GatewayMethodMetadata) methods.get("oneWay"); assertNotNull(meta); assertEquals("456", meta.getRequestTimeout()); assertEquals("123", meta.getReplyTimeout()); assertEquals("foo", meta.getReplyChannelName()); + meta = (GatewayMethodMetadata) methods.get("oneWayWithTimeouts"); + assertNotNull(meta); + assertEquals("#args[1]", meta.getRequestTimeout()); + assertEquals("#args[2]", meta.getReplyTimeout()); + service.oneWayWithTimeouts("foo", 100L, 200L); + result = channel.receive(10000); + assertNotNull(result); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml index 951073d286..669a469a56 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml @@ -19,7 +19,6 @@ - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java index 54f590a577..7b68129d31 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java @@ -46,7 +46,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; @@ -63,6 +62,7 @@ import org.springframework.context.annotation.FilterType; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.annotation.AnnotationConstants; import org.springframework.integration.annotation.BridgeTo; @@ -357,15 +357,17 @@ public class GatewayInterfaceTests { @Test public void testLateReply() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", + this.getClass()); Bar baz = ac.getBean(Bar.class); - String reply = baz.lateReply("hello"); + String reply = baz.lateReply("hello", 1000, 0); assertNull(reply); PollableChannel errorChannel = ac.getBean("errorChannel", PollableChannel.class); Message receive = errorChannel.receive(5000); assertNotNull(receive); MessagingException messagingException = (MessagingException) receive.getPayload(); - assertThat(messagingException.getMessage(), Matchers.startsWith("Reply message received but the receiving thread has exited due to a timeout")); + assertThat(messagingException.getMessage(), + startsWith("Reply message received but the receiving thread has exited due to a timeout")); ac.close(); } @@ -436,10 +438,12 @@ public class GatewayInterfaceTests { assertNotNull(this.gatewayByAnnotationGPFB); assertSame(this.exec, this.annotationGatewayProxyFactoryBean.getAsyncExecutor()); - assertEquals(1111L, - TestUtils.getPropertyValue(this.annotationGatewayProxyFactoryBean, "defaultRequestTimeout")); - assertEquals(222L, - TestUtils.getPropertyValue(this.annotationGatewayProxyFactoryBean, "defaultReplyTimeout")); + assertEquals(1111L, TestUtils + .getPropertyValue(this.annotationGatewayProxyFactoryBean, "defaultRequestTimeout", Expression.class) + .getValue()); + assertEquals(222L, TestUtils + .getPropertyValue(this.annotationGatewayProxyFactoryBean, "defaultReplyTimeout", Expression.class) + .getValue()); Collection messagingGateways = this.annotationGatewayProxyFactoryBean.getGateways().values(); @@ -490,7 +494,9 @@ public class GatewayInterfaceTests { void baz(String payload); - String lateReply(String payload); + @Gateway(payloadExpression = "#args[0]", requestChannel = "lateReplyChannel", + requestTimeoutExpression = "#args[1]", replyTimeoutExpression = "#args[2]") + String lateReply(String payload, long requestTimeout, long replyTimeout); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayXmlAndAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayXmlAndAnnotationTests.java index 62a8ef7fea..bd03fbeffa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayXmlAndAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayXmlAndAnnotationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -24,7 +24,9 @@ import java.util.Map.Entry; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.expression.Expression; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; @@ -44,7 +46,8 @@ public class GatewayXmlAndAnnotationTests { @Test public void test() { - assertEquals(123L, TestUtils.getPropertyValue(gatewayProxyFactoryBean, "defaultReplyTimeout")); + assertEquals(123L, TestUtils.getPropertyValue(gatewayProxyFactoryBean, "defaultReplyTimeout", Expression.class) + .getValue()); @SuppressWarnings("unchecked") Map gatewayMap = TestUtils.getPropertyValue(gatewayProxyFactoryBean, "gatewayMap", Map.class); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java index 55569ffa19..19e81b6f2e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java @@ -37,6 +37,9 @@ public interface TestService { void oneWay(String input); + @Payload("#args[0]") + void oneWayWithTimeouts(String input, Long sendTimeout, Long receiveTimeout); + String solicitResponse(); Message getMessage(); diff --git a/src/reference/asciidoc/endpoint-summary.adoc b/src/reference/asciidoc/endpoint-summary.adoc index bc571d5749..6c5d79a24f 100644 --- a/src/reference/asciidoc/endpoint-summary.adoc +++ b/src/reference/asciidoc/endpoint-summary.adoc @@ -420,6 +420,27 @@ To recap, *Inbound Channel Adapters* are used for one-way integration bringing d +| *STOMP* + + +| <> + + +| <> + + +| N + + +| N + + + + + + + + | *Stream* @@ -437,10 +458,6 @@ To recap, *Inbound Channel Adapters* are used for one-way integration bringing d - - - - | *Syslog* diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index 356da0de60..f03f7e7d26 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -430,6 +430,37 @@ sample demonstrates both techniques to return the exception to the caller. It emulates a Socket IO error to the waiting thread using an `aggregator` with `group-timeout` (see <>) and `MessagingTimeoutException` reply on the discard flow. +[[gateway-timeouts]] +==== Gateway Timeouts + +There are two properties `requestTimeout` and `replyTimeout`. +The request timeout only applies if the channel can block (e.g. a bounded `QueueChannel` that is full). +The reply timeout is how long the gateway will wait for a reply, or return `null`; it defaults to infinity. + +The timeouts can be set as defaults for all methods on the gateway (`defaultRequestTimeout`, `defaultReplyTimeout`) (or on the `MessagingGateway` interface annotation). +Individual methods can override these defaults (in `` child elements) or on the `@Gateway` annotation. + +Starting with _version 5.0_ the timeouts can be defined as expressions: + +[source, java] +---- +@Gateway(payloadExpression = "#args[0]", requestChannel = "someChannel", + requestTimeoutExpression = "#args[1]", replyTimeoutExpression = "#args[2]") +String lateReply(String payload, long requestTimeout, long replyTimeout); +---- + +The evaluation context has a `BeanResolver` (use `@someBean` to reference other beans) and the `#args` array variable is available. + +When configuring with XML, the timeout attributes can be a simple long value or a SpEL expression. + +[source, xml] +---- + + +---- [[async-gateway]] ==== Asynchronous Gateway diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 5703cab4b0..04ebbf74f9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -20,26 +20,31 @@ Also see the https://github.com/spring-projects/spring-integration/wiki/Spring-I A new Spring Integration Test Framework has been created to assist with testing Spring Integration applications. Now, with the `@SpringIntegrationTest` annotation on test class and `MockIntegration` factory you can make your JUnit tests for integration flows somewhat easier. + See <> for more information. ==== MongoDB Outbound Gateway The new `MongoDbOutboundGateway` allows you to make queries to the database on demand by sending a message to its request channel. + See <> for more information. ==== HTTP Reactive Outbound Gateway and Channel Adapter The new `ReactiveHttpRequestExecutingMessageHandler` adds support for WebFlux `WebClient` for outbound channel adapter and gateway. + See <> for more information. ==== Content Type Conversion Now that we use the new `InvocableHandlerMethod` -based infrastructure for service method invocations, we can perform `contentType` conversion from payload to target method argument. + See <> for more information. ==== ErrorMessagePublisher and ErrorMessageStrategy -The `ErrorMessagePublisher` abd the `ErrorMessageStrategy` are provided for creating `ErrorMessage` instances. +The `ErrorMessagePublisher` and the `ErrorMessageStrategy` are provided for creating `ErrorMessage` instances. + See <> for more information. [[x5.0-general]] @@ -51,25 +56,32 @@ Previous Project Reactor versions are no longer supported. ==== Core Changes The `@Poller` annotation now has the `errorChannel` attribute for easier configuration of the underlying `MessagePublishingErrorHandler`. + See <> for more information. All the request-reply endpoints (based on `AbstractReplyProducingMessageHandler`) can now start transaction and, therefore, make the whole downstream flow transactional. + See <> for more information. The `SmartLifecycleRoleController` now provides methods to obtain status of endpoints in roles. + See <> for more information. POJO methods are now invoked using an `InvocableHandlerMethod` by default, but can be configured to use SpEL as before. + See <> for more information. When targeting POJO methods as message handlers, one of the service methods can now be marked with the `@Default` annotation to provide a fallback mechanism for non-matched conditions. + See <> for more information. A simple `PassThroughTransactionSynchronizationFactory` is provided to always store a polled message in the current transaction context. That message is used as a `failedMessage` property of the `MessagingException` which wraps a raw exception thrown during transaction completion. + See <> for more information. The aggregator expression-based `ReleaseStrategy` now evaluates the expression against the `MesageGroup` instead of just the collection of `Message`. + See <> for more information. ==== Gateway Changes @@ -80,12 +92,15 @@ This had the effect that synchronous downstream flows (running on the calling th The `RequestReplyExchanger` interface now has a `throws MessagingException` clause to meet all the proposed messages exchange contract. -See <> for more information. +The request and reply timeouts can now be specified as SpEL expressions. + +See <> for more information. ==== Aggregator Performance Changes Aggregators now use a `SimpleSequenceSizeReleaseStrategy` by default, which is more efficient, especially with large groups. Empty groups are now scheduled for removal after `empty-group-min-timeout`. + See <> for more information. ==== Splitter Changes @@ -93,6 +108,7 @@ See <> for more information. The Splitter component now can handle and split Java `Stream` and Reactive Streams `Publisher` objects. If the output channel is a `ReactiveStreamsSubscribableChannel`, the `AbstractMessageSplitter` builds a `Flux` for subsequent iteration instead of a regular `Iterator` independent of object being split. In addition, `AbstractMessageSplitter` provides `protected obtainSizeIfPossible()` methods to allow the determination of the size of the `Iterable` and `Iterator` objects if that is possible. + See <> for more information. ==== JMS Changes @@ -107,33 +123,32 @@ See <> for more information. ==== Mail Changes Some inconsistencies with rendering IMAP mail content have been resolved. + See <> for more information. ==== Feed Changes Instead of the `com.rometools.fetcher.FeedFetcher`, which is deprecated in ROME, a new `Resource` property has been introduced to the `FeedEntryMessageSource`. + See <> for more information. ==== File Changes The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to represent relative path in the `FileReadingMessageSource`. -See <> for more information. The tail adapter now supports `idleEventInterval` to emit events when there is no data in the file during that period. -See <> for more information. The flush predicates for the `FileWritingMessageHandler` now have an additional parameter. -See <> for more information. The file outbound channel adapter and gateway (`FileWritingMessageHandler`) now support the `REPLACE_IF_MODIFIED` `FileExistsMode`. -See <> for more information. They also now support setting file permissions on the newly written file. -See <> for more information. A new `FileSystemMarkerFilePresentFileListFilter` is now available; see <> for more information. +See <> for more information. + ==== (S)FTP Changes The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory. @@ -159,43 +174,45 @@ The `FtpOutboundGateway` can now be supplied with `workingDirExpression` to chan The `RemoteFileTemplate` is supplied now with the `invoke(OperationsCallback action)` to perform several `RemoteFileOperations` calls in the scope of the same, thread-bounded, `Session`. -See <> and <> for more information. +New filters for detecting incomplete remote files are now provided. -New filters for detecting incomplete remote files are now provided; see <> and <> for more information. +See <> and <> for more information. ==== Integration Properties Since _version 4.3.2_ a new `spring.integration.readOnly.headers` global property has been added to customize the list of headers which should not be copied to a newly created `Message` by the `MessageBuilder`. + See <> for more information. ==== Stream Changes There is a new option on the `CharacterStreamReadingMessageSource` to allow it to be used to "pipe" stdin and publish an application event when the pipe is closed. + See <> for more information. ==== Barrier Changes The `BarrierMessageHandler` now supports a discard channel to which late-arriving trigger messages are sent. + See <> for more information. ==== AMQP Changes The AMQP outbound endpoints now support setting a delay expression for when using the RabbitMQ Delayed Message Exchange plugin. -See <> for more information. The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`. -See <> for more information. Pollable AMQP-backed channels now block the poller thread for the poller's configured `receiveTimeout` (default 1 second). -See <> for more information. Headers, such as `contentType` that are added to message properties by the message converter are now used in the final message; previously, it depended on the converter type as to which headers/message properties appeared in the final message. To override headers set by the converter, set the `headersMappedLast` property to `true`. -See <> for more information. + +See <> for more information. ==== HTTP Changes The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`. + See <> for more information. ==== MQTT Changes @@ -208,7 +225,7 @@ See <> for more information. ==== STOMP Changes -The STOMP module has been changed to use `ReactorNettyTcpStompClient`, based on the Project Reactor `3.0` and `reactor-netty` extension. +The STOMP module has been changed to use `ReactorNettyTcpStompClient`, based on the Project Reactor `3.1` and `reactor-netty` extension. The `Reactor2TcpStompSessionManager` has been renamed to the `ReactorNettyTcpStompSessionManager` according to the `ReactorNettyTcpStompClient` foundation. See <> for more information. @@ -233,9 +250,10 @@ See <> for more information. ==== TCP Changes - A new `ThreadAffinityClientConnectionFactory` is provided that binds TCP connections to threads. -See <> for more information. You can now configure the TCP connection factories to support `PushbackInputStream` s, allowing deserializers to "unread" (push back) bytes after "reading ahead". -See <> for more information. + +A `ByteArrayElasticRawDeserializer` has been added without `maxMessageSize` control and buffer incoming data as needed. + +See <> for more information.