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

INT-4525 Configure TCP gateway second chance delay

Browse Source 
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
This commit is contained in:
Gary Russell
2018-09-05 12:55:44 -04:00
committed by Artem Bilan
parent 47a74abd30
commit 6cc4fe530d
2 changed files with 91 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String, AsyncReply> pendingReplies = new ConcurrentHashMap<String, AsyncReply>();
private final Map<String, AsyncReply> 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) {

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

@@ -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<Socket> 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<String, ?> 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<String> 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();
}
}