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

INT-2861/INT-2862 TCP Outbound Gateway Fixes

Browse Source 
INT-2861

Close connection after 'remoteTimeout' because the
socket is dirty (may contain an in-flight reply).

Add test that demonstrates the problem and that it
is resolved.

INT-2862

Remove entries from pendingReplies (map of async responses
for which we are waiting).

Add an assertion to the above test to ensure cleanup.
This commit is contained in:
Gary Russell
2012-12-13 12:38:14 +05:30
committed by Mark Fisher
parent e66a098149
commit 90132686ae
3 changed files with 148 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -51,11 +51,11 @@ import org.springframework.util.Assert;
*/
public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener, SmartLifecycle {
private volatile AbstractConnectionFactory connectionFactory;
private volatile AbstractClientConnectionFactory connectionFactory;
private Map<String, AsyncReply> pendingReplies = new ConcurrentHashMap<String, AsyncReply>();
private final Map<String, AsyncReply> pendingReplies = new ConcurrentHashMap<String, AsyncReply>();
private Semaphore semaphore = new Semaphore(1, true);
private final Semaphore semaphore = new Semaphore(1, true);
private volatile long remoteTimeout = 10000L;
@@ -100,6 +100,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
Assert.notNull(connectionFactory, this.getClass().getName() +
" requires a client connection factory");
boolean haveSemaphore = false;
String connectionId = null;
try {
boolean singleUseConnection = this.connectionFactory.isSingleUse();
if (!singleUseConnection) {
@@ -114,13 +115,19 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
TcpConnection connection = this.connectionFactory.getConnection();
AsyncReply reply = new AsyncReply();
pendingReplies.put(connection.getConnectionId(), reply);
connectionId = connection.getConnectionId();
pendingReplies.put(connectionId, reply);
if (logger.isDebugEnabled()) {
logger.debug("Added " + connection.getConnectionId());
}
connection.send(requestMessage);
Message<?> replyMessage = reply.getReply();
if (replyMessage == null) {
if (logger.isDebugEnabled()) {
logger.debug("Remote Timeout on " + connection.getConnectionId());
}
// The connection is dirty - force it closed.
this.connectionFactory.forceClose(connection);
throw new MessageTimeoutException(requestMessage, "Timed out waiting for response");
}
if (logger.isDebugEnabled()) {
@@ -136,6 +143,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
throw new MessagingException("Failed to send or receive", e);
}
finally {
if (connectionId != null) {
pendingReplies.remove(connectionId);
}
if (haveSemaphore) {
this.semaphore.release();
if (logger.isDebugEnabled()) {
@@ -161,9 +171,10 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
// TODO: In 3.0 Change parameter type to AbstractClientConnectionFactory
Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory,
this.getClass().getName() + " requires a client connection factory");
this.connectionFactory = connectionFactory;
this.connectionFactory = (AbstractClientConnectionFactory) connectionFactory;
connectionFactory.registerListener(this);
connectionFactory.registerSender(this);
}

12
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java
View File

@@ -96,4 +96,16 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
return theConnection;
}
/**
* Force close the connection and null the field if it's
* a shared connection.
* @param connection
*/
public void forceClose(TcpConnection connection) {
if (this.theConnection == connection) {
this.theConnection = null;
}
connection.close();
}
}

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

@@ -21,11 +21,15 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
@@ -34,9 +38,12 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ServerSocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
@@ -55,6 +62,8 @@ import org.springframework.integration.test.util.TestUtils;
*/
public class TcpOutboundGatewayTests {
private final Log logger = LogFactory.getLog(this.getClass());
@Test
public void testGoodNetSingle() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
@@ -223,8 +232,6 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 2; i++) {
try {
results[i].get();
} catch (InterruptedException e) {
} catch (ExecutionException e) {
if (timeouts > 0) {
fail("Unexpected " + e.getMessage());
@@ -248,5 +255,116 @@ public class TcpOutboundGatewayTests {
done.set(true);
}
/**
* Sends 2 concurrent messages on a shared connection. The GW single threads
* these requests. The first will timeout; the second should receive its
* own response, not that for the first.
* @throws Exception
*/
@Test
public void testGoodNetGWTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
/*
* The payload of the last message received by the remote side;
* used to verify the correct response is received.
*/
final AtomicReference<String> lastReceived = new AtomicReference<String>();
final CountDownLatch serverLatch = new CountDownLatch(2);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
int i = 0;
while (!done.get()) {
Socket socket = server.accept();
i++;
while (!socket.isClosed()) {
try {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
String request = (String) ois.readObject();
logger.debug("Read " + request);
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
if (i < 2) {
Thread.sleep(1000);
}
oos.writeObject(request.replace("Test", "Reply"));
logger.debug("Replied to " + request);
lastReceived.set(request);
serverLatch.countDown();
}
catch (IOException e) {
socket.close();
}
}
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.setSingleUse(false);
ccf.start();
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
final TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(ccf);
gateway.setRequestTimeout(Integer.MAX_VALUE);
QueueChannel replyChannel = new QueueChannel();
gateway.setRequiresReply(true);
gateway.setOutputChannel(replyChannel);
gateway.setRemoteTimeout(500);
@SuppressWarnings("unchecked")
Future<Integer>[] results = new Future[2];
for (int i = 0; i < 2; i++) {
final int j = i;
results[j] = (Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
public Integer call() throws Exception {
// increase the timeout after the first send
if (j > 0) {
gateway.setRemoteTimeout(5000);
}
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
return j;
}
}));
Thread.sleep(50);
}
// wait until the server side has processed both requests
assertTrue(serverLatch.await(10, TimeUnit.SECONDS));
List<String> replies = new ArrayList<String>();
int timeouts = 0;
for (int i = 0; i < 2; i++) {
try {
int result = results[i].get();
String reply = (String) replyChannel.receive(1000).getPayload();
logger.debug(i + " got " + result + " " + reply);
replies.add(reply);
} catch (ExecutionException e) {
if (timeouts >= 2) {
fail("Unexpected " + e.getMessage());
} else {
assertNotNull(e.getCause());
assertTrue(e.getCause() instanceof MessageTimeoutException);
}
timeouts++;
continue;
}
}
assertEquals("Expected exactly one ExecutionException", 1, timeouts);
assertEquals(1, replies.size());
assertEquals(lastReceived.get().replace("Test", "Reply"), replies.get(0));
done.set(true);
assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size());
}
}