From 3133782cc5dae745d55363d65633ac81bb39f7f8 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 13 Aug 2015 15:10:48 -0400 Subject: [PATCH] INT-3801: TCP Server Fix NPE with Early Stop JIRA: https://jira.spring.io/browse/INT-3801 NPE if the server is stopped before it fully started. Also fix SOLinger tests. Fix `ConnectionFactoryTests` for Java < 8 compatibility --- .../connection/AbstractConnectionFactory.java | 2 +- .../TcpNetServerConnectionFactory.java | 16 +++- .../TcpNioServerConnectionFactory.java | 20 +++-- .../connection/ConnectionFactoryTests.java | 79 +++++++++++++++++++ .../tcp/connection/SOLingerTests-context.xml | 4 +- .../ip/tcp/connection/SOLingerTests.java | 51 +++++++----- 6 files changed, 140 insertions(+), 32 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 11c84b5f28..62cbcbe67a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -892,7 +892,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport public String toString() { return super.toString() + (this.host != null ? ", host=" + this.host : "") - + ", port=" + this.port; + + ", port=" + getPort(); } private class PendingIO { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index b7f40fc853..57087e00cd 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -85,7 +85,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto public void run() { ServerSocket theServerSocket = null; if (getListener() == null) { - logger.info("No listener bound to server connection factory; will not read; exiting..."); + logger.info(this + " No listener bound to server connection factory; will not read; exiting..."); return; } try { @@ -99,7 +99,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto getTcpSocketSupport().postProcessServerSocket(theServerSocket); this.serverSocket = theServerSocket; setListening(true); - logger.info("Listening on port " + getPort()); + logger.info(this + " Listening"); while (true) { final Socket socket; /* @@ -107,7 +107,15 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto * Not fatal. */ try { - socket = serverSocket.accept(); + if (this.serverSocket == null) { + if (logger.isDebugEnabled()) { + logger.debug(this + " stopped before accept"); + } + throw new IOException(this + " stopped before accept"); + } + else { + socket = this.serverSocket.accept(); + } } catch (SocketTimeoutException ste) { if (logger.isDebugEnabled()) { @@ -140,7 +148,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto catch (Exception e) { // don't log an error if we had a good socket once and now it's closed if (e instanceof SocketException && theServerSocket != null) { - logger.warn("Server Socket closed"); + logger.info("Server Socket closed"); } else if (isActive()) { logger.error("Error on ServerSocket; port = " + getPort(), e); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 2a8ef46898..1ad999dd92 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -105,7 +105,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto @Override public void run() { if (getListener() == null) { - logger.info("No listener bound to server connection factory; will not read; exiting..."); + logger.info(this + " No listener bound to server connection factory; will not read; exiting..."); return; } try { @@ -121,14 +121,20 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port), Math.abs(getBacklog())); } if (logger.isInfoEnabled()) { - logger.info("Listening on port " + getPort()); + logger.info(this + " Listening"); } final Selector selector = Selector.open(); - this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); - setListening(true); - this.selector = selector; - doSelect(this.serverChannel, selector); - + if (this.serverChannel == null) { + if (logger.isDebugEnabled()) { + logger.debug(this + " stopped before registering the server channel"); + } + } + else { + this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); + setListening(true); + this.selector = selector; + doSelect(this.serverChannel, selector); + } } catch (IOException e) { if (isActive()) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java index a2d7bc46c9..f664467218 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java @@ -16,13 +16,19 @@ package org.springframework.integration.ip.tcp.connection; +import static org.hamcrest.Matchers.hasItem; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Matchers.any; +import static org.mockito.Matchers.contains; +import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -30,16 +36,21 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; import org.springframework.integration.ip.util.TestingUtilities; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; /** @@ -143,6 +154,74 @@ public class ConnectionFactoryTests { serverFactory.stop(); } + @Test + public void testEarlyCloseNet() throws Exception { + AbstractServerConnectionFactory factory = new TcpNetServerConnectionFactory(0); + testEarlyClose(factory, "serverSocket", " stopped before accept"); + } + + @Test + public void testEarlyCloseNio() throws Exception { + AbstractServerConnectionFactory factory = new TcpNioServerConnectionFactory(0); + testEarlyClose(factory, "serverChannel", " stopped before registering the server channel"); + } + + private void testEarlyClose(final AbstractServerConnectionFactory factory, String property, + String message) throws Exception { + factory.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); + factory.setBeanName("foo"); + factory.registerListener(mock(TcpListener.class)); + factory.afterPropertiesSet(); + Log logger = spy(TestUtils.getPropertyValue(factory, "logger", Log.class)); + new DirectFieldAccessor(factory).setPropertyValue("logger", logger); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + when(logger.isInfoEnabled()).thenReturn(true); + when(logger.isDebugEnabled()).thenReturn(true); + doAnswer(new Answer() { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + latch1.countDown(); + // wait until the stop nulls the channel + latch2.await(10, TimeUnit.SECONDS); + return null; + } + }).when(logger).info(contains("Listening")); + doAnswer(new Answer() { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + latch3.countDown(); + return null; + } + }).when(logger).debug(contains(message)); + factory.start(); + assertTrue("missing info log", latch1.await(10, TimeUnit.SECONDS)); + // stop on a different thread because it waits for the executor + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + factory.stop(); + } + }); + int n = 0; + DirectFieldAccessor accessor = new DirectFieldAccessor(factory); + while (n++ < 200 && accessor.getPropertyValue(property) != null) { + Thread.sleep(100); + } + assertTrue("Stop was not invoked in time", n < 200); + latch2.countDown(); + assertTrue("missing debug log", latch3.await(10, TimeUnit.SECONDS)); + String expected = "foo, port=" + factory.getPort() + message; + ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); + verify(logger, atLeast(1)).debug(captor.capture()); + assertThat(captor.getAllValues(), hasItem(expected)); + factory.stop(); + } + @SuppressWarnings("serial") private class FooEvent extends TcpConnectionOpenEvent { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml index 559ad2ef0d..0bc82ee363 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml @@ -58,7 +58,7 @@ port="#{tcpIpUtils.findAvailableServerSocket(9400)}" so-timeout="1000" single-use="true" - so-linger="1000" + so-linger="10000" />