diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
index 872328a533..085636095b 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
@@ -125,7 +125,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.support.AnnotationConfigContextLoader;
import reactor.Environment;
-import reactor.fn.Consumer;
import reactor.rx.Promise;
import reactor.rx.Streams;
import reactor.spring.context.config.EnableReactor;
@@ -930,6 +929,7 @@ public class EnableIntegrationTests {
@Override
protected boolean doSend(Message message, long timeout) {
+ logger.debug("---- 'publishedChannel' Thread State: " + Thread.currentThread().isInterrupted());
logger.debug("---- 'publishedChannel' before 'doSend': " + message);
logger.debug("---- 'publishedChannel' state before: " + getQueueSize() +
", " + getRemainingCapacity());
diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
index 1caaa6767e..8f06249a43 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
@@ -98,6 +98,8 @@ public abstract class IpAdapterParserUtils {
public static final String REMOTE_TIMEOUT = "remote-timeout";
+ public static final String REMOTE_TIMEOUT_EXPRESSION = "remote-timeout-expression";
+
public static final String REPLY_TIMEOUT = "reply-timeout";
public static final String REPLY_CHANNEL = "reply-channel";
diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
index e50b736ae8..a07f8a8d90 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.ip.config;
import org.w3c.dom.Element;
+import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
@@ -46,8 +47,12 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IpAdapterParserUtils.REPLY_CHANNEL);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.REQUEST_TIMEOUT);
- IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
- IpAdapterParserUtils.REMOTE_TIMEOUT);
+ BeanDefinition remoteTimeoutExpression = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression
+ (IpAdapterParserUtils.REMOTE_TIMEOUT, IpAdapterParserUtils.REMOTE_TIMEOUT_EXPRESSION,
+ parserContext, element, false);
+ if (remoteTimeoutExpression != null) {
+ builder.addPropertyValue("remoteTimeoutExpression", remoteTimeoutExpression);
+ }
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.REPLY_TIMEOUT, "sendTimeout");
return builder;
diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
index e5ed9691ed..7b8ef93105 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2001-2014 the original author or authors.
+ * Copyright 2001-2015 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,12 @@ import java.util.concurrent.TimeUnit;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
+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.MessageTimeoutException;
+import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -52,7 +57,7 @@ import org.springframework.util.Assert;
* @since 2.0
*/
public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
- implements TcpSender, TcpListener, Lifecycle {
+ implements TcpSender, TcpListener, IntegrationEvaluationContextAware, Lifecycle {
private volatile AbstractClientConnectionFactory connectionFactory;
@@ -60,12 +65,12 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
private final Semaphore semaphore = new Semaphore(1, true);
- private volatile long remoteTimeout = 10000L;
-
- private volatile boolean remoteTimeoutSet = false;
+ private volatile Expression remoteTimeoutExpression = new LiteralExpression("10000");
private volatile long requestTimeout = 10000;
+ private volatile EvaluationContext evaluationContext = new StandardEvaluationContext();
+
/**
* @param requestTimeout the requestTimeout to set
*/
@@ -77,21 +82,19 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
* @param remoteTimeout the remoteTimeout to set
*/
public void setRemoteTimeout(long remoteTimeout) {
- this.remoteTimeout = remoteTimeout;
- this.remoteTimeoutSet = true;
+ this.remoteTimeoutExpression = new LiteralExpression("" + remoteTimeout);
+ }
+
+ /**
+ * @param remoteTimeoutExpression the remoteTimeoutExpression to set
+ */
+ public void setRemoteTimeoutExpression(Expression remoteTimeoutExpression) {
+ this.remoteTimeoutExpression = remoteTimeoutExpression;
}
@Override
- public void setSendTimeout(long sendTimeout) {
- super.setSendTimeout(sendTimeout);
- /*
- * For backwards compatibility, also set the remote
- * timeout to this value, unless it has been
- * explicitly set.
- */
- if (!this.remoteTimeoutSet) {
- this.remoteTimeout = sendTimeout;
- }
+ public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
+ this.evaluationContext = evaluationContext;
}
@Override
@@ -113,7 +116,8 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
}
}
TcpConnection connection = this.connectionFactory.getConnection();
- AsyncReply reply = new AsyncReply();
+ AsyncReply reply = new AsyncReply(this.remoteTimeoutExpression.getValue(this.evaluationContext,
+ requestMessage, Long.class));
connectionId = connection.getConnectionId();
pendingReplies.put(connectionId, reply);
if (logger.isDebugEnabled()) {
@@ -249,11 +253,14 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
private final CountDownLatch secondChanceLatch;
+ private final long remoteTimeout;
+
private volatile Message reply;
- public AsyncReply() {
+ public AsyncReply(long remoteTimeout) {
this.latch = new CountDownLatch(1);
this.secondChanceLatch = new CountDownLatch(1);
+ this.remoteTimeout = remoteTimeout;
}
/**
@@ -263,7 +270,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
*/
public Message getReply() throws Exception {
try {
- if (!this.latch.await(remoteTimeout, TimeUnit.MILLISECONDS)) {
+ if (!this.latch.await(this.remoteTimeout, TimeUnit.MILLISECONDS)) {
return null;
}
}
diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.2.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.2.xsd
index 436268611c..02d2c0da5a 100644
--- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.2.xsd
+++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.2.xsd
@@ -402,7 +402,18 @@
from the remote system. Prior to 2.2, this was specified
with the reply-timeout attribute. To provide easier migration,
this attribute defaults to the same value of the reply-timeout,
- if supplied, or 10 seconds otherwise.
+ if supplied, or 10 seconds otherwise. Mutually exclusive with
+ 'remote-timeout-expression'.
+
+
+
+
+
+
+ Specifies an expresssion that is evaluated against the outbound message
+ to determine the time the gateway will wait for a reply
+ from the remote system. Mutually exclusive with
+ 'remote-timeout'.
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
index e1d1a500af..b18a079239 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
@@ -234,6 +234,7 @@
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
index c951164452..be56377780 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
@@ -491,11 +491,13 @@ public class ParserUnitTests {
MessagingTemplate messagingTemplate = TestUtils.getPropertyValue(tcpOutboundGateway, "messagingTemplate",
MessagingTemplate.class);
assertEquals(Long.valueOf(567), TestUtils.getPropertyValue(messagingTemplate, "sendTimeout", Long.class));
- assertEquals(789L, dfa.getPropertyValue("remoteTimeout"));
+ assertEquals("789", TestUtils.getPropertyValue(tcpOutboundGateway, "remoteTimeoutExpression.literalValue"));
assertEquals("outGateway",tcpOutboundGateway.getComponentName());
assertEquals("ip:tcp-outbound-gateway", tcpOutboundGateway.getComponentType());
assertTrue(cfC2.isLookupHost());
assertEquals(24, dfa.getPropertyValue("order"));
+
+ assertEquals("4000", TestUtils.getPropertyValue(outAdviceGateway, "remoteTimeoutExpression.expression"));
}
@Test
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
index ea8c78e374..a4a7d38a1c 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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,6 +57,7 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -125,14 +126,12 @@ public class TcpOutboundGatewayTests {
gateway.setRequiresReply(true);
gateway.setOutputChannel(replyChannel);
// check the default remote timeout
- assertEquals(Long.valueOf(10000), TestUtils.getPropertyValue(gateway, "remoteTimeout", Long.class));
+ assertEquals("10000", TestUtils.getPropertyValue(gateway, "remoteTimeoutExpression.literalValue"));
gateway.setSendTimeout(123);
- // ensure this also changed the remote timeout
- assertEquals(Long.valueOf(123), TestUtils.getPropertyValue(gateway, "remoteTimeout", Long.class));
gateway.setRemoteTimeout(60000);
gateway.setSendTimeout(61000);
// ensure this did NOT change the remote timeout
- assertEquals(Long.valueOf(60000), TestUtils.getPropertyValue(gateway, "remoteTimeout", Long.class));
+ assertEquals("60000", TestUtils.getPropertyValue(gateway, "remoteTimeoutExpression.literalValue"));
gateway.setRequestTimeout(60000);
for (int i = 100; i < 200; i++) {
gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build());
@@ -685,7 +684,7 @@ public class TcpOutboundGatewayTests {
QueueChannel replyChannel = new QueueChannel();
gateway.setRequiresReply(true);
gateway.setOutputChannel(replyChannel);
- gateway.setRemoteTimeout(5000);
+ gateway.setRemoteTimeoutExpression(new SpelExpressionParser().parseExpression("5000"));
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet();
gateway.start();
@@ -702,4 +701,5 @@ public class TcpOutboundGatewayTests {
done.set(true);
ccf.getConnection();
}
+
}
diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml
index 513ee48411..6b26b5015e 100644
--- a/src/reference/docbook/ip.xml
+++ b/src/reference/docbook/ip.xml
@@ -709,7 +709,7 @@
reply-channel="replyChannel"
connection-factory="cfClient"
request-timeout="10000"
- remote-timeout="10000"/>]]>
+ remote-timeout="10000"/> ]]>
A simple outbound TCP gateway.
@@ -1960,16 +1960,22 @@ private CompositeExecutor compositeExecutor() {
reply-channel
- Optional. The channel to which reply messages may be sent if the
- original outbound message did not contain a reply channel header.
-
+ Optional. The channel to which reply messages may be sent.
remote-timeout
The time in milliseconds for which the gateway will wait for a reply from the
- remote system.
- Default: Same value as reply-timeout, if specified, or 10000 (10 seconds) otherwise.
+ remote system. Mutually exclusive with remote-timeout-expression.
+ Default: 10000 (10 seconds). Note: in versions prior to 4.2
+ this value defaulted to reply-timeout (if set).
+
+
+ remote-timeout-expression
+
+ A SpEL expression, evaluated against the message to determine
+ the time in milliseconds for which the gateway will wait for a reply from the
+ remote system. Mutually exclusive with remote-timeout.
request-timeout
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml
index 1de9830652..b555fa7ecd 100644
--- a/src/reference/docbook/whats-new.xml
+++ b/src/reference/docbook/whats-new.xml
@@ -56,6 +56,21 @@
See for more information.
+
+ TCP Gateway Remote Timeout
+
+ The TcpOutboundGateway now supports remote-timeout-expression
+ as an alternative to the existing remote-timeout attribute.
+ This allows setting the timeout based on each message.
+
+
+ Also, the remote-timeout no longer defaults to the same value as reply-timeout
+ which has a completely different meaning.
+
+
+ See for more information.
+
+