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

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 {

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);

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()) {

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<Void>() {
+
+			@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<Void>() {
+
+			@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<String> 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 {
 

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"
 		/>
 
 	<int-ip:tcp-inbound-gateway request-channel="echo"
@@ -70,7 +70,7 @@
 		so-timeout="1000"
 		single-use="true"
 		using-nio="true"
-		so-linger="1000"
+		so-linger="10000"
 		/>
 
 	<int-ip:tcp-inbound-gateway request-channel="echo"

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java

@@ -16,7 +16,6 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.io.IOException;
@@ -26,7 +25,6 @@ import java.net.SocketException;
 
 import javax.net.SocketFactory;
 
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -68,14 +66,13 @@ public class SOLingerTests {
 	public void configOk() {}
 
 	@Test
-	@Ignore
 	public void finReceivedNet() {
-		finReceived(inCFNet);
+		finReceived(inCFNet, false);
 	}
 
 	@Test
 	public void finReceivedNio() {
-		finReceived(inCFNio);
+		finReceived(inCFNio, false);
 	}
 
 	@Test
@@ -90,16 +87,19 @@ public class SOLingerTests {
 
 	@Test
 	public void finReceivedNetLinger() {
-		finReceived(inCFNetLinger);
+		finReceived(inCFNetLinger, true);
 	}
 
 	@Test
-	@Ignore
 	public void finReceivedNioLinger() {
-		finReceived(inCFNioLinger);
+		finReceived(inCFNioLinger, true);
 	}
 
-	private void finReceived(AbstractServerConnectionFactory inCF) {
+	private void finReceived(AbstractServerConnectionFactory inCF, boolean hasLinger) {
+		/*
+		 * Default (no linger) means the OS may still deliver everything before the
+		 * FIN, but it's not guaranteed.
+		 */
 		int port = inCF.getPort();
 		TestingUtilities.waitListening(inCF, null);
 		try {
@@ -108,12 +108,23 @@ public class SOLingerTests {
 			String test = "Test\r\n";
 			socket.getOutputStream().write(test.getBytes());
 			byte[] buff = new byte[test.length() + 5];
-			readFully(socket.getInputStream(), buff);
-			assertEquals("echo:" + test, new String(buff));
+			try {
+				readFully(socket.getInputStream(), buff);
+				assertEquals("echo:" + test, new String(buff));
+			}
+			catch (SocketException se) {
+				if (hasLinger) {
+					fail("SocketException not expected with SO_LINGER");
+				}
+				else {
+					return;
+				}
+			}
 			int n = socket.getInputStream().read();
 			// we expect an orderly close
 			assertEquals(-1, n);
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			e.printStackTrace();
 			fail("Unexpected Exception  " + e.getMessage());
 		}
@@ -129,15 +140,19 @@ public class SOLingerTests {
 			String test = "Test\r\n";
 			socket.getOutputStream().write(test.getBytes());
 			byte[] buff = new byte[test.length() + 5];
-			readFully(socket.getInputStream(), buff);
-			assertEquals("echo:" + test, new String(buff));
 			try {
+				// with SO_LINGER=0 we may, or may not, get the data
+				// if we do, verify it is as expected, if not, the RST
+				// arrived before the final data.
+				readFully(socket.getInputStream(), buff);
+				assertEquals("echo:" + test, new String(buff));
 				socket.getInputStream().read();
-				fail("Expected IOException");
-			} catch (IOException ioe) {
-				assertTrue(ioe instanceof SocketException);
+				fail("Expected SocketException");
 			}
-		} catch (Exception e) {
+			catch (SocketException se) {
+			}
+		}
+		catch (Exception e) {
 			e.printStackTrace();
 			fail("Unexpected Exception  " + e.getMessage());
 		}