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 1845bd9d3d..7a20e3ecc3 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 @@ -57,14 +57,19 @@ import org.apache.commons.logging.LogFactory; import org.apache.log4j.Level; import org.junit.Rule; import org.junit.Test; +import org.mockito.Matchers; import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; +import org.springframework.expression.EvaluationContext; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory; @@ -383,26 +388,41 @@ public class TcpOutboundGatewayTests { QueueChannel replyChannel = new QueueChannel(); gateway.setRequiresReply(true); gateway.setOutputChannel(replyChannel); - gateway.setRemoteTimeout(500); + + ValueExpression remoteTimeoutExpression = Mockito.spy(new ValueExpression(500L)); + + final AtomicBoolean remoteTimeoutUsed = new AtomicBoolean(); + + Mockito.doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + if (!remoteTimeoutUsed.getAndSet(true)) { + // increase the timeout after the first send + gateway.setRemoteTimeout(5000); + } + return invocation.callRealMethod(); + } + + }).when(remoteTimeoutExpression) + .getValue(Mockito.any(EvaluationContext.class), Matchers.any()); + + gateway.setRemoteTimeoutExpression(remoteTimeoutExpression); @SuppressWarnings("unchecked") Future[] results = (Future[]) new Future[2]; + final CountDownLatch secondMessageLatch = new CountDownLatch(1); for (int i = 0; i < 2; i++) { final int j = i; results[j] = (Executors.newSingleThreadExecutor().submit(new Callable() { + @Override public Integer call() throws Exception { - try { - gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); - } - finally { - // increase the timeout after the first send - if (j > 0) { - gateway.setRemoteTimeout(5000); - } - } + gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); return j; } + })); + } // wait until the server side has processed both requests assertTrue(serverLatch.await(10, TimeUnit.SECONDS)); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java index 7ff1063d70..74f11d07d9 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.jms; import static org.junit.Assert.assertEquals; @@ -64,8 +65,8 @@ import org.springframework.util.ObjectUtils; /** * @author Gary Russell + * @author Artem Bilan * @since 2.2.4 - * */ public class JmsOutboundGatewayTests { @@ -81,8 +82,8 @@ public class JmsOutboundGatewayTests { gateway.setBeanFactory(mock(BeanFactory.class)); gateway.afterPropertiesSet(); assertEquals("JMS_OutboundGateway@" + ObjectUtils.getIdentityHexString(gateway) + - ".replyListener", - TestUtils.getPropertyValue(gateway, "replyContainer.beanName")); + ".replyListener", + TestUtils.getPropertyValue(gateway, "replyContainer.beanName")); } @Test @@ -94,15 +95,17 @@ public class JmsOutboundGatewayTests { gateway.setUseReplyContainer(true); ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties(); final List errors = new ArrayList(); - ErrorHandlingTaskExecutor errorHandlingTaskExecutor = new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() { + ErrorHandlingTaskExecutor errorHandlingTaskExecutor = + new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() { - @Override - public void handleError(Throwable t) { - logger.info("Error:", t); - errors.add(t); - throw new RuntimeException(t); - } - }); + @Override + public void handleError(Throwable t) { + logger.info("Error:", t); + errors.add(t); + throw new RuntimeException(t); + } + + }); replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor); replyContainerProperties.setRecoveryInterval(100L); gateway.setReplyContainerProperties(replyContainerProperties); @@ -115,7 +118,9 @@ public class JmsOutboundGatewayTests { public Connection answer(InvocationOnMock invocation) throws Throwable { int theCount = connectionAttempts.incrementAndGet(); if (theCount > 1 && theCount < 4) { - throw new JmsException("bar") {}; + throw new JmsException("bar") { + + }; } return connection; } @@ -134,7 +139,9 @@ public class JmsOutboundGatewayTests { public Message answer(InvocationOnMock invocation) throws Throwable { int theCount = count.incrementAndGet(); if (theCount > 1 && theCount < 4) { - throw new JmsException("foo") {}; + throw new JmsException("foo") { + + }; } if (theCount > 4) { Thread.sleep(100); @@ -190,7 +197,7 @@ public class JmsOutboundGatewayTests { CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); JmsTemplate template = new JmsTemplate(connectionFactory2); - template.setReceiveTimeout(5000); + template.setReceiveTimeout(10000); template.afterPropertiesSet(); final Message request = template.receive(requestQ); assertNotNull(request); @@ -205,7 +212,7 @@ public class JmsOutboundGatewayTests { } }; template.send(replyQ, reply); - org.springframework.messaging.Message received = queueChannel.receive(10000); + org.springframework.messaging.Message received = queueChannel.receive(20000); assertNotNull(received); assertEquals("bar", received.getPayload()); gateway.stop(); @@ -240,7 +247,7 @@ public class JmsOutboundGatewayTests { CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); JmsTemplate template = new JmsTemplate(connectionFactory2); - template.setReceiveTimeout(5000); + template.setReceiveTimeout(10000); template.afterPropertiesSet(); final Message request = template.receive(requestQ); assertNotNull(request); @@ -255,7 +262,7 @@ public class JmsOutboundGatewayTests { } }; template.send(replyQ, reply); - org.springframework.messaging.Message received = queueChannel.receive(10000); + org.springframework.messaging.Message received = queueChannel.receive(20000); assertNotNull(received); assertEquals("bar", received.getPayload()); connectionFactory1.destroy();