From 6cc4fe530df1d7ee3617bfdb5d2ba628ff568c92 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 5 Sep 2018 12:55:44 -0400 Subject: [PATCH] INT-4525 Configure TCP gateway second chance delay JIRA: https://jira.spring.io/browse/INT-4525 Hard-coded 2 second delay is now configurable. Also fix the logic for creating the SpEL evaluation context. **cherry-pick to 5.0.x** * Polishing according PR comments --- .../ip/tcp/TcpOutboundGateway.java | 44 ++++++++++---- .../ip/tcp/TcpOutboundGatewayTests.java | 59 +++++++++++++++++++ 2 files changed, 91 insertions(+), 12 deletions(-) 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 16e7df447c..6ca9f20442 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-2017 the original author or authors. + * Copyright 2001-2018 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. @@ -56,24 +56,31 @@ import org.springframework.util.Assert; * * * @author Gary Russell + * * @since 2.0 */ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener, Lifecycle { - private volatile AbstractClientConnectionFactory connectionFactory; + private static final int DEFAULT_SECOND_CHANCE_DELAY = 2; - private volatile boolean isSingleUse; - - private final Map pendingReplies = new ConcurrentHashMap(); + private final Map pendingReplies = new ConcurrentHashMap<>(); private final Semaphore semaphore = new Semaphore(1, true); - private volatile Expression remoteTimeoutExpression = new LiteralExpression("10000"); + private AbstractClientConnectionFactory connectionFactory; - private volatile long requestTimeout = 10000; + private boolean isSingleUse; - private volatile EvaluationContext evaluationContext = new StandardEvaluationContext(); + private Expression remoteTimeoutExpression = new LiteralExpression("10000"); + + private long requestTimeout = 10000; + + private EvaluationContext evaluationContext = new StandardEvaluationContext(); + + private boolean evaluationContextSet; + + private int secondChanceDelay = DEFAULT_SECOND_CHANCE_DELAY; /** * @param requestTimeout the requestTimeout to set @@ -97,18 +104,31 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler } public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + Assert.notNull(evaluationContext, "'evaluationContext' cannot be null"); this.evaluationContext = evaluationContext; + this.evaluationContextSet = true; } @Override protected void doInit() { super.doInit(); - - if (this.evaluationContext == null) { + if (!this.evaluationContextSet) { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); } } + /** + * When using NIO and the server closes the socket after sending the reply, + * an error message representing the close may appear before the reply. + * Set the delay, in seconds, to wait for an actual reply after an {@link ErrorMessage} is + * received. Default 2 seconds. + * @param secondChanceDelay the delay. + * @since 5.0.8 + */ + public void setSecondChanceDelay(int secondChanceDelay) { + this.secondChanceDelay = secondChanceDelay; + } + @Override protected Object handleRequestMessage(Message requestMessage) { Assert.notNull(this.connectionFactory, this.getClass().getName() + @@ -178,7 +198,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler @Override public boolean onMessage(Message message) { - String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + String connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class); if (connectionId == null) { logger.error("Cannot correlate response - no connection id"); publishNoConnectionEvent(message, null, "Cannot correlate response - no connection id"); @@ -322,7 +342,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler * before the reply, on a different thread. */ logger.debug("second chance"); - this.secondChanceLatch.await(2, TimeUnit.SECONDS); // NOSONAR don't care about result + this.secondChanceLatch.await(TcpOutboundGateway.this.secondChanceDelay, TimeUnit.SECONDS); // NOSONAR waitForMessageAfterError = false; } else if (this.reply.getPayload() instanceof MessagingException) { 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 d9fec545ec..b2b136df09 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 @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp; +import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -67,6 +68,7 @@ import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory; @@ -79,6 +81,7 @@ import org.springframework.integration.test.support.LongRunningIntegrationTest; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; /** @@ -850,4 +853,60 @@ public class TcpOutboundGatewayTests { ccf.stop(); } + @Test + public void testNioSecondChance() throws Exception { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); + final int port = server.getLocalPort(); + TcpOutboundGateway gateway = new TcpOutboundGateway(); + TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", port); + cf.setApplicationEventPublisher(e -> { }); + gateway.setConnectionFactory(cf); + final AtomicBoolean done = new AtomicBoolean(); + + this.executor.execute(() -> { + List sockets = new ArrayList<>(); + try { + while (!done.get()) { + sockets.add(server.accept()); + } + } + catch (Exception e1) { + if (!done.get()) { + e1.printStackTrace(); + } + } + for (Socket socket : sockets) { + try { + socket.close(); + } + catch (IOException e2) { + } + } + }); + QueueChannel replies = new QueueChannel(); + gateway.setReplyChannel(replies); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setRemoteTimeout(60_000); + gateway.afterPropertiesSet(); + gateway.start(); + this.executor.execute(() -> gateway.handleMessage(new GenericMessage<>("foo"))); + int n = 0; + @SuppressWarnings("unchecked") + Map pending = TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class); + while (n++ < 100 && pending.size() == 0) { + Thread.sleep(100); + } + assertTrue(pending.size() > 0); + String connectionId = pending.keySet().iterator().next(); + this.executor.execute(() -> gateway.onMessage(new ErrorMessage(new RuntimeException(), + Collections.singletonMap(IpHeaders.CONNECTION_ID, connectionId)))); + GenericMessage message = new GenericMessage<>("FOO", + Collections.singletonMap(IpHeaders.CONNECTION_ID, connectionId)); + gateway.onMessage(message); + assertThat(replies.receive(10000), equalTo(message)); + gateway.stop(); + done.set(true); + server.close(); + } + }