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

TCP ConnectionFactoryTests Race Condition

Browse Source 
Wait for events.
This commit is contained in:
Gary Russell
2015-12-21 11:38:55 -05:00
parent 0458220756
commit ed874ea84a
2 changed files with 17 additions and 7 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

@@ -376,6 +376,9 @@ public abstract class TcpConnectionSupport implements TcpConnection {
} }
else { else {
this.applicationEventPublisher.publishEvent(event); this.applicationEventPublisher.publishEvent(event);
if (logger.isTraceEnabled()) {
logger.trace("Published: " + event);
}
} }
} }
catch (Exception e) { catch (Exception e) {

21
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java
View File

@@ -43,6 +43,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test; import org.junit.Test;
import org.mockito.ArgumentCaptor; import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock; import org.mockito.invocation.InvocationOnMock;
@@ -55,6 +56,7 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.ip.event.IpIntegrationEvent; import org.springframework.integration.ip.event.IpIntegrationEvent;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.springframework.integration.test.support.LogAdjustingTestSupport;
import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message; import org.springframework.messaging.Message;
import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.TaskScheduler;
@@ -66,7 +68,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
* @since 3.0 * @since 3.0
* *
*/ */
public class ConnectionFactoryTests { public class ConnectionFactoryTests extends LogAdjustingTestSupport {
@Test @Test
public void testObtainConnectionIdsNet() throws Exception { public void testObtainConnectionIdsNet() throws Exception {
@@ -83,15 +85,21 @@ public class ConnectionFactoryTests {
public void testObtainConnectionIds(AbstractServerConnectionFactory serverFactory) throws Exception { public void testObtainConnectionIds(AbstractServerConnectionFactory serverFactory) throws Exception {
final List<IpIntegrationEvent> events = final List<IpIntegrationEvent> events =
Collections.synchronizedList(new ArrayList<IpIntegrationEvent>()); Collections.synchronizedList(new ArrayList<IpIntegrationEvent>());
int expectedEvents = serverFactory instanceof TcpNetServerConnectionFactory
? 7 // Listening, + OPEN, CLOSE, EXCEPTION for each side
: 5; // Listening, + OPEN, CLOSE (but we *might* get exceptions, depending on timing).
final CountDownLatch serverListeningLatch = new CountDownLatch(1); final CountDownLatch serverListeningLatch = new CountDownLatch(1);
final CountDownLatch eventLatch = new CountDownLatch(expectedEvents);
ApplicationEventPublisher publisher = new ApplicationEventPublisher() { ApplicationEventPublisher publisher = new ApplicationEventPublisher() {
@Override @Override
public void publishEvent(ApplicationEvent event) { public void publishEvent(ApplicationEvent event) {
LogFactory.getLog(this.getClass()).trace("Received: " + event);
events.add((IpIntegrationEvent) event); events.add((IpIntegrationEvent) event);
if (event instanceof TcpConnectionServerListeningEvent) { if (event instanceof TcpConnectionServerListeningEvent) {
serverListeningLatch.countDown(); serverListeningLatch.countDown();
} }
eventLatch.countDown();
} }
@Override @Override
@@ -153,15 +161,14 @@ public class ConnectionFactoryTests {
clients = clientFactory.getOpenConnectionIds(); clients = clientFactory.getOpenConnectionIds();
} }
assertEquals(0, clients.size()); assertEquals(0, clients.size());
int expected = serverFactory instanceof TcpNetServerConnectionFactory ? 7// Listening, OPEN, CLOSE, EXCEPTION for each side assertTrue(eventLatch.await(10, TimeUnit.SECONDS));
: 5; //Listening, OPEN, CLOSE (but we *might* get exceptions, depending on timing). assertThat("Expected at least " + expectedEvents + " events; got: " + events.size() + " : " + events,
assertThat("Expected at least " + expected + " events; got: " + events.size() + " : " + events, events.size(), greaterThanOrEqualTo(expectedEvents));
events.size(), greaterThanOrEqualTo(expected));
FooEvent event = new FooEvent(client, "foo"); FooEvent event = new FooEvent(client, "foo");
client.publishEvent(event); client.publishEvent(event);
assertThat("Expected at least " + expected + " events; got: " + events.size() + " : " + events, assertThat("Expected at least " + expectedEvents + " events; got: " + events.size() + " : " + events,
events.size(), greaterThanOrEqualTo(expected + 1)); events.size(), greaterThanOrEqualTo(expectedEvents + 1));
try { try {
event = new FooEvent(mock(TcpConnectionSupport.class), "foo"); event = new FooEvent(mock(TcpConnectionSupport.class), "foo");