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

INT-4033: Support SpEL in Gateway Timeouts

Browse Source 
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
This commit is contained in:
Gary Russell
2017-05-23 21:28:52 -04:00
committed by Artem Bilan
parent 914094e51d
commit f58106ec3c
20 changed files with 421 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-integration-core/src/main/java/org/springframework/integration/annotation/Gateway.java
View File

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

19
spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java
View File

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

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

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

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

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

38
spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java
View File

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

6
spring-integration-core/src/main/java/org/springframework/integration/expression/ValueExpression.java
View File

@@ -30,6 +30,7 @@ import org.springframework.util.Assert;
* @param <V> - The expected value type.
*
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
public class ValueExpression<V> implements Expression {
@@ -169,4 +170,9 @@ public class ValueExpression<V> implements Expression {
return this.value.toString();
}
@Override
public String toString() {
return "ValueExpression [value=" + this.value + "]";
}
}

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

@@ -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<Object[]>, 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<Object[]
private final MethodArgsMessageMapper argsMapper;
private final MessageBuilderFactory messageBuilderFactory;
private volatile Expression payloadExpression;
private final Map<String, Expression> parameterPayloadExpressions = new HashMap<String, Expression>();
@@ -105,7 +109,9 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
private volatile BeanFactory beanFactory;
private final MessageBuilderFactory messageBuilderFactory;
private Expression sendTimeoutExpression;
private Expression replyTimeoutExpression;
GatewayMethodInboundMessageMapper(Method method) {
this(method, null);
@@ -152,11 +158,17 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
}
@Override
public void setBeanFactory(final BeanFactory beanFactory) {
if (beanFactory != null) {
this.beanFactory = beanFactory;
this.payloadExpressionEvaluationContext = ExpressionUtils.createStandardEvaluationContext(beanFactory);
}
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.payloadExpressionEvaluationContext = ExpressionUtils.createStandardEvaluationContext(beanFactory);
}
public void setSendTimeoutExpression(Expression sendTimeoutExpression) {
this.sendTimeoutExpression = sendTimeoutExpression;
}
public void setReplyTimeoutExpression(Expression replyTimeoutExpression) {
this.replyTimeoutExpression = replyTimeoutExpression;
}
@Override
@@ -186,12 +198,10 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
private Map<String, Object> evaluateHeaders(EvaluationContext methodInvocationEvaluationContext,
Map<String, Expression> headerExpressions) {
Map<String, Object> evaluatedHeaders = new HashMap<String, Object>();
Map<String, Object> evaluatedHeaders = new HashMap<>();
for (Map.Entry<String, Expression> 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<Object[]
}
private Object evaluatePayloadExpression(String expressionString, Object argumentValue) {
Expression expression = this.parameterPayloadExpressions.get(expressionString);
if (expression == null) {
expression = PARSER.parseExpression(expressionString);
this.parameterPayloadExpressions.put(expressionString, expression);
}
Expression expression =
this.parameterPayloadExpressions.computeIfAbsent(expressionString, PARSER::parseExpression);
return expression.getValue(this.payloadExpressionEvaluationContext, argumentValue);
}
@@ -218,8 +225,8 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
for (Entry<?, ?> 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<Object[]
private void throwExceptionForMultipleMessageOrPayloadParameters(MethodParameter methodParameter) {
throw new MessagingException(
"At most one parameter (or expression via method-level @Payload) may be mapped to the " +
"payload or Message. Found more than one on method [" + methodParameter.getMethod() + "]");
"payload or Message. Found more than one on method [" + methodParameter.getMethod() + "]");
}
private String determineHeaderName(Annotation headerAnnotation, MethodParameter methodParameter) {
@@ -244,7 +251,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
}
private static List<MethodParameter> getMethodParameterList(Method method) {
List<MethodParameter> parameterList = new LinkedList<MethodParameter>();
List<MethodParameter> 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<Object[]
GatewayMethodInboundMessageMapper.this.copyHeaders((Map<?, ?>) 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)) {

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

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

16
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

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

14
spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd
View File

@@ -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.
]]>
</xsd:documentation>
</xsd:annotation>
@@ -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.
]]>
</xsd:documentation>
</xsd:annotation>
@@ -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.
]]>
</xsd:documentation>
</xsd:annotation>
@@ -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.
]]>
</xsd:documentation>
</xsd:annotation>

7
spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java
View File

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

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

@@ -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">
<default-header name="baz" value="qux"/>
<method name="oneWay" request-channel="otherRequestChannel"
@@ -88,6 +90,10 @@
reply-channel="foo">
<header name="foo" value="bar"/>
</method>
<method name="oneWayWithTimeouts" request-channel="otherRequestChannel"
request-timeout="#args[1]"
reply-timeout="#args[2]">
</method>
</gateway>
<!-- no assertions for this. The fact that this config does not result in error is sufficient -->

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

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

1
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml
View File

@@ -19,7 +19,6 @@
<int:method name="baz">
<int:header name="name" value="overrideGlobal"/>
</int:method>
<int:method name="lateReply" request-channel="lateReplyChannel" reply-timeout="0"/>
</int:gateway>
<int:chain input-channel="lateReplyChannel" >

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

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

7
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayXmlAndAnnotationTests.java
View File

@@ -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<Method, MessagingGatewaySupport> gatewayMap = TestUtils.getPropertyValue(gatewayProxyFactoryBean,
"gatewayMap", Map.class);

3
spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java
View File

@@ -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<String> getMessage();

25
src/reference/asciidoc/endpoint-summary.adoc
View File

@@ -420,6 +420,27 @@ To recap, *Inbound Channel Adapters* are used for one-way integration bringing d
| *STOMP*
| <<stomp-inbound-adapter>>
| <<stomp-outbound-adapter>>
| N
| N
| *Stream*
@@ -437,10 +458,6 @@ To recap, *Inbound Channel Adapters* are used for one-way integration bringing d
| *Syslog*

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

@@ -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 <<agg-and-group-to>>)
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 `<method/>` 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]
----
<method name="someMethod" request-channel="someRequestChannel"
payload-expression="#args[0]"
request-timeout="1000"
reply-timeout="#args[1]">
</method>
----
[[async-gateway]]
==== Asynchronous Gateway

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

@@ -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 <<testing>> 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 <<mongodb-outbound-gateway>> 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 <<http-outbound>> 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 <<content-type-conversion>> 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 <<namespace-errorhandler>> 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 <<annotations>> for more information.
All the request-reply endpoints (based on `AbstractReplyProducingMessageHandler`) can now start transaction and, therefore, make the whole downstream flow transactional.
See <<tx-handle-message-advice>> for more information.
The `SmartLifecycleRoleController` now provides methods to obtain status of endpoints in roles.
See <<endpoint-roles>> for more information.
POJO methods are now invoked using an `InvocableHandlerMethod` by default, but can be configured to use SpEL as before.
See <<pojo-invocation>> 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 <<service-activator-namespace>> 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 <<transaction-synchronization>> for more information.
The aggregator expression-based `ReleaseStrategy` now evaluates the expression against the `MesageGroup` instead of just the collection of `Message<?>`.
See <<aggregator-spel>> 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 <<gateway-error-handling>> for more information.
The request and reply timeouts can now be specified as SpEL expressions.
See <<gateway>> 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 <<aggregator>> for more information.
==== Splitter Changes
@@ -93,6 +108,7 @@ See <<aggregator>> 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 <<splitter>> for more information.
==== JMS Changes
@@ -107,33 +123,32 @@ See <<jms>> for more information.
==== Mail Changes
Some inconsistencies with rendering IMAP mail content have been resolved.
See <<imap-format-important, the note in the Mail-Receiving Channel Adapter Section>> 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 <<feed>> for more information.
==== File Changes
The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to represent relative path in the `FileReadingMessageSource`.
See <<file-reading>> for more information.
The tail adapter now supports `idleEventInterval` to emit events when there is no data in the file during that period.
See <<file-tailing>> for more information.
The flush predicates for the `FileWritingMessageHandler` now have an additional parameter.
See <<file-flushing>> for more information.
The file outbound channel adapter and gateway (`FileWritingMessageHandler`) now support the `REPLACE_IF_MODIFIED` `FileExistsMode`.
See <<file-writing-destination-exists>> for more information.
They also now support setting file permissions on the newly written file.
See <<file-permissions>> for more information.
A new `FileSystemMarkerFilePresentFileListFilter` is now available; see <<file-incomplete>> for more information.
See <<files>> 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<F, T> action)` to perform several `RemoteFileOperations` calls in the scope of the same, thread-bounded, `Session`.
See <<ftp>> and <<sftp>> for more information.
New filters for detecting incomplete remote files are now provided.
New filters for detecting incomplete remote files are now provided; see <<ftp-incomplete>> and <<sftp-incomplete>> for more information.
See <<ftp>> and <<sftp>> 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 <<global-properties>> 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 <<stream-reading>> for more information.
==== Barrier Changes
The `BarrierMessageHandler` now supports a discard channel to which late-arriving trigger messages are sent.
See <<barrier>> 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 <<amqp-delay>> for more information.
The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`.
See <<amqp-inbound-channel-adapter>> for more information.
Pollable AMQP-backed channels now block the poller thread for the poller's configured `receiveTimeout` (default 1 second).
See <<amqp-channels>> 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 <<content-type-conversion-outbound>> for more information.
See <<amqp>> for more information.
==== HTTP Changes
The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`.
See <<http-header-mapping>> for more information.
==== MQTT Changes
@@ -208,7 +225,7 @@ See <<mqtt>> 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 <<stomp>> for more information.
@@ -233,9 +250,10 @@ See <<redis>> for more information.
==== TCP Changes
A new `ThreadAffinityClientConnectionFactory` is provided that binds TCP connections to threads.
See <<tcp-affinity-cf>> 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 <<tcp-advanced-techniques>> for more information.
A `ByteArrayElasticRawDeserializer` has been added without `maxMessageSize` control and buffer incoming data as needed.
See <<ip>> for more information.