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

INT-2620 Fix Race Condition in Failover Tests

Browse Source 
Sporadic test failures (fairly consistently on Mac).

Problem was some tests sent a message to a client socket before the
close notification had been received.

Wait for client connection to close rather than waiting for the
server to stop.

Also, with NIO, close the channel on accept exception.

Also improve debugging by including a log of new connections
including the id.
This commit is contained in:
Gary Russell
2012-06-26 08:00:33 -04:00
parent 7d6b5850f5
commit bf4005a92c
4 changed files with 42 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
View File

@@ -120,6 +120,9 @@ public abstract class TcpConnectionSupport implements TcpConnection {
this.connectionFactoryName = connectionFactoryName;
}
this.publishConnectionOpenEvent();
if (logger.isDebugEnabled()) {
logger.debug("New connection " + this);
}
}
public void afterSend(Message<?> message) throws Exception {

26
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
View File

@@ -156,17 +156,23 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
channel.close();
}
else {
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
if (connection == null) {
return;
try {
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
if (connection == null) {
return;
}
connection.setTaskExecutor(this.getTaskExecutor());
connection.setLastRead(now);
this.channelMap.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
}
catch (Exception e) {
logger.error("Exception accepting new connection", e);
channel.close();
}
connection.setTaskExecutor(this.getTaskExecutor());
connection.setLastRead(now);
this.channelMap.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
}
}

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -15,6 +15,7 @@
*/
package org.springframework.integration.ip.util;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
/**
@@ -88,4 +89,23 @@ public class TestingUtilities {
}
}
/**
* Wait for up to 10 seconds for the connection factory to have the specified number
* of connections.
* @param factory The factory.
* @param n The required number of connections.
* @throws Exception IllegalStateException if the count does not match.
*/
public static void waitUntilFactoryHasThisNumberOfConnections(AbstractConnectionFactory factory, int n)
throws Exception {
int timer = 0;
while (timer < 10000) {
if (factory.getOpenConnectionIds().size() == n) {
return;
}
Thread.sleep(100);
timer += 100;
}
throw new IllegalStateException("Connections=" + factory.getOpenConnectionIds().size() + "wanted=" + n);
}
}

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java
View File

@@ -335,7 +335,7 @@ public class FailoverClientConnectionFactoryTests {
Message<?> replyMessage = replyChannel.receive(10000);
assertNotNull(replyMessage);
server1.stop();
TestingUtilities.waitStopListening(server1, null);
TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(client1, 0);
outGateway.handleMessage(message);
socket = getSocket(client2);
port2 = socket.getLocalPort();
@@ -343,6 +343,7 @@ public class FailoverClientConnectionFactoryTests {
replyMessage = replyChannel.receive(10000);
assertNotNull(replyMessage);
gateway2.stop();
outGateway.stop();
}
private Socket getSocket(AbstractClientConnectionFactory client) throws Exception {