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

INT-3635: TCP GW: Add remoteTimeoutExpression

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3635

Add an expression as an alternative to `remote-timeout`.

INT-3635: Tcp GW Docs

Also removed the relationship between reply-timeout and
remote-timeout (introduced in 2.2 to provide backward compatibility
after the incorrect use of the reply-timeout attribute).

* Add log for Thread interruption state to the `EnableIntegrationTests`: https://build.spring.io/browse/INT-B41-JOB1-233
This commit is contained in:
Gary Russell
2015-03-02 13:39:43 +02:00
committed by Artem Bilan
parent 80a45715ac
commit 40c4cb0d41
10 changed files with 87 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

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

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
View File

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

47
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

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

13
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.2.xsd
View File

@@ -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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="remote-timeout-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

1
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -234,6 +234,7 @@
<ip:tcp-outbound-gateway id="outAdviceGateway"
request-channel="tcpAdviceGateChannel"
reply-channel="replyChannel"
remote-timeout-expression="4000"
connection-factory="mockClientCf">
<ip:request-handler-advice-chain>
<bean class="org.springframework.integration.ip.config.ParserUnitTests$FooAdvice" />

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

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

12
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

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

18
src/reference/docbook/ip.xml
View File

@@ -709,7 +709,7 @@
reply-channel="replyChannel"
connection-factory="cfClient"
request-timeout="10000"
remote-timeout="10000"/>]]></programlisting>
remote-timeout="10000"/> <!-- or e.g. remote-timeout-expression="headers['timeout']" -->]]></programlisting>
A simple outbound TCP gateway.
</para>
</section>
@@ -1960,16 +1960,22 @@ private CompositeExecutor compositeExecutor() {
<row>
<entry>reply-channel</entry>
<entry></entry>
<entry>Optional. The channel to which reply messages may be sent if the
original outbound message did not contain a reply channel header.
</entry>
<entry>Optional. The channel to which reply messages may be sent.</entry>
</row>
<row>
<entry>remote-timeout</entry>
<entry></entry>
<entry>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.</entry>
remote system. Mutually exclusive with <code>remote-timeout-expression</code>.
Default: 10000 (10 seconds). Note: in versions prior to <emphasis>4.2</emphasis>
this value defaulted to <code>reply-timeout</code> (if set).</entry>
</row>
<row>
<entry>remote-timeout-expression</entry>
<entry></entry>
<entry>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 <code>remote-timeout</code>.</entry>
</row>
<row>
<entry>request-timeout</entry>

15
src/reference/docbook/whats-new.xml
View File

@@ -56,6 +56,21 @@
See <xref linkend="tcp-events"/> for more information.
</para>
</section>
<section id="4.2-tcp-gw-rto">
<title>TCP Gateway Remote Timeout</title>
<para>
The <classname>TcpOutboundGateway</classname> now supports <code>remote-timeout-expression</code>
as an alternative to the existing <code>remote-timeout</code> attribute.
This allows setting the timeout based on each message.
</para>
<para>
Also, the <code>remote-timeout</code> no longer defaults to the same value as <code>reply-timeout</code>
which has a completely different meaning. <!-- MIGRATION -->
</para>
<para>
See <xref linkend="tcp-ob-gateway-attributes"/> for more information.
</para>
</section>
<section id="4.2-inbound-channel-adapter-annotation">
<title>@InboundChannelAdapter</title>
<para>