From 40c4cb0d417599169af8d143852dd46965f059e4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 2 Mar 2015 13:39:43 +0200 Subject: [PATCH] INT-3635: TCP GW: Add remoteTimeoutExpression 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 --- .../configuration/EnableIntegrationTests.java | 2 +- .../ip/config/IpAdapterParserUtils.java | 2 + .../ip/config/TcpOutboundGatewayParser.java | 11 +++-- .../ip/tcp/TcpOutboundGateway.java | 47 +++++++++++-------- .../ip/config/spring-integration-ip-4.2.xsd | 13 ++++- .../ip/config/ParserUnitTests-context.xml | 1 + .../ip/config/ParserUnitTests.java | 4 +- .../ip/tcp/TcpOutboundGatewayTests.java | 12 ++--- src/reference/docbook/ip.xml | 18 ++++--- src/reference/docbook/whats-new.xml | 15 ++++++ 10 files changed, 87 insertions(+), 38 deletions(-) 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. + +
@InboundChannelAdapter