INT-3646: Add TCP Server Exception Events

JIRA: https://jira.spring.io/browse/INT-3646

Publish `TcpConnectionServerExceptionEvent`s when unexpected
exceptions occur on server sockets.

INT-3646: Polishing
Fix Reactor tests

spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.gateway;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -84,7 +85,7 @@ public class AsyncGatewayTests {
 		Object result = f.get(1000, TimeUnit.MILLISECONDS);
 		long elapsed = System.currentTimeMillis() - start;
 		assertTrue(elapsed >= 200);
-		assertTrue(result instanceof Message<?>);
+		assertNotNull(result);
 		assertEquals("foobar", ((Message<?>) result).getPayload());
 	}
 
@@ -149,7 +150,6 @@ public class AsyncGatewayTests {
 		long elapsed = System.currentTimeMillis() - start;
 		assertTrue(elapsed >= 200);
 		assertEquals("foobar", result.get().getPayload());
-
 		Object thread = result.get().getHeaders().get("thread");
 		assertNotEquals(Thread.currentThread(), thread);
 	}
@@ -169,7 +169,6 @@ public class AsyncGatewayTests {
 		CustomFuture f = service.returnCustomFuture("foo");
 		String result = f.get(1000, TimeUnit.MILLISECONDS);
 		assertEquals("foobar", result);
-
 		assertEquals(Thread.currentThread(), f.thread);
 	}
 
@@ -184,14 +183,13 @@ public class AsyncGatewayTests {
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 
-		proxyFactory.setAsyncExecutor(null);	// Not async - user flow returns Future<?>
+		proxyFactory.setAsyncExecutor(null);    // Not async - user flow returns Future<?>
 
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		CustomFuture f = (CustomFuture) service.returnCustomFutureWithTypeFuture("foo");
 		String result = f.get(1000, TimeUnit.MILLISECONDS);
 		assertEquals("foobar", result);
-
 		assertEquals(Thread.currentThread(), f.thread);
 	}
 
@@ -204,6 +202,7 @@ public class AsyncGatewayTests {
 						.setHeader("thread", Thread.currentThread())
 						.build();
 			}
+
 		});
 	}
 
@@ -222,9 +221,8 @@ public class AsyncGatewayTests {
 		long start = System.currentTimeMillis();
 		Object result = f.get(1000, TimeUnit.MILLISECONDS);
 		long elapsed = System.currentTimeMillis() - start;
-
 		assertTrue(elapsed >= 200 - safety);
-		assertTrue(result instanceof String);
+		assertNotNull(result);
 		assertEquals("foobar", result);
 	}
 
@@ -243,7 +241,6 @@ public class AsyncGatewayTests {
 		long start = System.currentTimeMillis();
 		Object result = f.get(1000, TimeUnit.MILLISECONDS);
 		long elapsed = System.currentTimeMillis() - start;
-
 		assertTrue(elapsed >= 200 - safety);
 		assertTrue(result instanceof String);
 		assertEquals("foobar", result);
@@ -263,10 +260,7 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Promise<Message<?>> promise = service.returnMessagePromise("foo");
-		long start = System.currentTimeMillis();
 		Object result = promise.await(1, TimeUnit.SECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-		assertTrue(elapsed <= 200);
 		assertEquals("foobar", ((Message<?>) result).getPayload());
 	}
 
@@ -283,11 +277,7 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Promise<String> promise = service.returnStringPromise("foo");
-		long start = System.currentTimeMillis();
 		Object result = promise.await(1, TimeUnit.SECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-
-		assertTrue(elapsed <= 200 - safety);
 		assertEquals("foobar", result);
 	}
 
@@ -304,12 +294,8 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Promise<?> promise = service.returnSomethingPromise("foo");
-		long start = System.currentTimeMillis();
 		Object result = promise.await(1, TimeUnit.SECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-
-		assertTrue(elapsed <= 200 - safety);
-		assertTrue(result instanceof String);
+		assertNotNull(result);
 		assertEquals("foobar", result);
 	}
 
@@ -326,7 +312,6 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Promise<String> promise = service.returnStringPromise("foo");
-		long start = System.currentTimeMillis();
 
 		final AtomicReference<String> result = new AtomicReference<String>();
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -340,9 +325,6 @@ public class AsyncGatewayTests {
 		});
 
 		latch.await(1, TimeUnit.SECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-
-		assertTrue(elapsed <= 200 - safety);
 		assertEquals("foobar", result.get());
 	}
 
@@ -366,6 +348,7 @@ public class AsyncGatewayTests {
 
 	private static void startResponder(final PollableChannel requestChannel) {
 		new Thread(new Runnable() {
+
 			@Override
 			public void run() {
 				Message<?> input = requestChannel.receive();
@@ -389,11 +372,12 @@ public class AsyncGatewayTests {
 				}
 				((MessageChannel) input.getHeaders().getReplyChannel()).send(reply);
 			}
+
 		}).start();
 	}
 
 
-	static interface TestEchoService {
+	interface TestEchoService {
 
 		Future<String> returnString(String s);
 
@@ -403,10 +387,10 @@ public class AsyncGatewayTests {
 
 		ListenableFuture<Message<?>> returnMessageListenable(String s);
 
-		@Gateway(headers=@GatewayHeader(name="method", expression="#gatewayMethod.name"))
+		@Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name"))
 		CustomFuture returnCustomFuture(String s);
 
-		@Gateway(headers=@GatewayHeader(name="method", expression="#gatewayMethod.name"))
+		@Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name"))
 		Future<?> returnCustomFutureWithTypeFuture(String s);
 
 		Promise<String> returnStringPromise(String s);

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2014 the original author or authors.
+ * Copyright 2001-2015 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.
@@ -57,10 +57,10 @@ public abstract class AbstractServerConnectionFactory
 	@Override
 	public void start() {
 		synchronized (this.lifecycleMonitor) {
-			if (!this.isActive()) {
+			if (!isActive()) {
 				this.setActive(true);
 				this.shuttingDown = false;
-				this.getTaskExecutor().execute(this);
+				getTaskExecutor().execute(this);
 			}
 		}
 		super.start();
@@ -102,22 +102,22 @@ public abstract class AbstractServerConnectionFactory
 	 * @param socket The new socket.
 	 */
 	protected void initializeConnection(TcpConnectionSupport connection, Socket socket) {
-		TcpListener listener = this.getListener();
+		TcpListener listener = getListener();
 		if (listener != null) {
 			connection.registerListener(listener);
 		}
-		connection.registerSender(this.getSender());
-		connection.setMapper(this.getMapper());
-		connection.setDeserializer(this.getDeserializer());
-		connection.setSerializer(this.getSerializer());
-		connection.setSingleUse(this.isSingleUse());
+		connection.registerSender(getSender());
+		connection.setMapper(getMapper());
+		connection.setDeserializer(getDeserializer());
+		connection.setSerializer(getSerializer());
+		connection.setSingleUse(isSingleUse());
 		/*
 		 * If we are configured
 		 * for single use; need to enforce a timeout on the socket so we will close
 		 * if the client connects, but sends nothing. (Protect against DoS).
 		 * Behavior can be overridden by explicitly setting the timeout to zero.
 		 */
-		if (this.isSingleUse() && this.getSoTimeout() < 0) {
+		if (isSingleUse() && getSoTimeout() < 0) {
 			try {
 				socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);
 			} catch (SocketException e) {
@@ -128,7 +128,7 @@ public abstract class AbstractServerConnectionFactory
 	}
 
 	protected void postProcessServerSocket(ServerSocket serverSocket) {
-		this.getTcpSocketSupport().postProcessServerSocket(serverSocket);
+		getTcpSocketSupport().postProcessServerSocket(serverSocket);
 	}
 
 	/**
@@ -154,7 +154,7 @@ public abstract class AbstractServerConnectionFactory
 	 * @return The backlog.
 	 */
 	public int getBacklog() {
-		return backlog;
+		return this.backlog;
 	}
 
 	/**
@@ -175,8 +175,12 @@ public abstract class AbstractServerConnectionFactory
 
 	@Override
 	public int afterShutdown() {
-		this.stop();
+		stop();
 		return 0;
 	}
 
+	protected void publishServerExceptionEvent(Exception e) {
+		getApplicationEventPublisher().publishEvent(new TcpConnectionServerExceptionEvent(this, e));
+	}
+
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerExceptionEvent.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.ip.tcp.connection;
+
+import org.springframework.integration.ip.event.IpIntegrationEvent;
+import org.springframework.util.Assert;
+
+/**
+ * {@link IpIntegrationEvent} representing exceptions on a TCP server socket/channel.
+ *
+ * @author Gary Russell
+ * @since 4.0.7
+ */
+@SuppressWarnings("serial")
+public class TcpConnectionServerExceptionEvent extends IpIntegrationEvent {
+
+	public TcpConnectionServerExceptionEvent(AbstractServerConnectionFactory connectionFactory, Throwable cause) {
+		super(connectionFactory, cause);
+		Assert.notNull(cause, "'cause' cannot be null");
+		Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
+	}
+
+	/**
+	 * The connection factory that experienced the exception; examine it to determine the port etc.
+	 * @return the connection factory.
+	 */
+	public AbstractServerConnectionFactory getConnectionFactory() {
+		return (AbstractServerConnectionFactory) getSource();
+	}
+
+	@Override
+	public String toString() {
+		return super.toString() +
+				", [factory=" + getConnectionFactory().getComponentType() +
+				":" + getConnectionFactory().getComponentName() +
+				", port=" + getConnectionFactory().getPort() + "]";
+	}
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -30,6 +30,7 @@ import org.springframework.util.Assert;
 /**
  * Implements a server connection factory that produces {@link TcpNetConnection}s using
  * a {@link ServerSocket}. Must have a {@link TcpListener} registered.
+ *
  * @author Gary Russell
  * @since 2.0
  *
@@ -48,6 +49,11 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		super(port);
 	}
 
+	@Override
+	public String getComponentType() {
+		return "tcp-net-server-connection-factory";
+	}
+
 	/**
 	 * If no listener registers, exits.
 	 * Accepts incoming connections and creates TcpConnections for each new connection.
@@ -58,22 +64,22 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 	@Override
 	public void run() {
 		ServerSocket theServerSocket = null;
-		if (this.getListener() == null) {
+		if (getListener() == null) {
 			logger.info("No listener bound to server connection factory; will not read; exiting...");
 			return;
 		}
 		try {
-			if (this.getLocalAddress() == null) {
-				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), null);
+			if (getLocalAddress() == null) {
+				theServerSocket = createServerSocket(getPort(), getBacklog(), null);
 			}
 			else {
-				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
-				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), whichNic);
+				InetAddress whichNic = InetAddress.getByName(getLocalAddress());
+				theServerSocket = createServerSocket(getPort(), getBacklog(), whichNic);
 			}
-			this.getTcpSocketSupport().postProcessServerSocket(theServerSocket);
+			getTcpSocketSupport().postProcessServerSocket(theServerSocket);
 			this.serverSocket = theServerSocket;
-			this.setListening(true);
-			logger.info("Listening on port " + this.getPort());
+			setListening(true);
+			logger.info("Listening on port " + getPort());
 			while (true) {
 				final Socket socket;
 				/*
@@ -89,7 +95,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 					}
 					continue;
 				}
-				if (this.isShuttingDown()) {
+				if (isShuttingDown()) {
 					if (logger.isInfoEnabled()) {
 						logger.info("New connection from " + socket.getInetAddress().getHostAddress()
 								+ " rejected; the server is in the process of shutting down.");
@@ -101,12 +107,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 						logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
 					}
 					setSocketAttributes(socket);
-					TcpConnectionSupport connection = new TcpNetConnection(socket, true, this.isLookupHost(),
-							this.getApplicationEventPublisher(), this.getComponentName());
+					TcpConnectionSupport connection = new TcpNetConnection(socket, true, isLookupHost(),
+							getApplicationEventPublisher(), getComponentName());
 					connection = wrapConnection(connection);
-					this.initializeConnection(connection, socket);
-					this.getTaskExecutor().execute(connection);
-					this.harvestClosedConnections();
+					initializeConnection(connection, socket);
+					getTaskExecutor().execute(connection);
+					harvestClosedConnections();
 					connection.publishConnectionOpenEvent();
 				}
 			}
@@ -115,20 +121,21 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 			// 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");
-			} else if (this.isActive()) {
-				logger.error("Error on ServerSocket", e);
+			}
+			else if (isActive()) {
+				logger.error("Error on ServerSocket; port = " + getPort(), e);
+				publishServerExceptionEvent(e);
 			}
 		}
 		finally {
-			this.setListening(false);
-			this.setActive(false);
+			setListening(false);
+			setActive(false);
 		}
 	}
 
 	/**
 	 * Create a new {@link ServerSocket}. This default implementation uses the default
 	 * {@link ServerSocketFactory}. Override to use some other mechanism
-	 *
 	 * @param port The port.
 	 * @param backlog The server socket backlog.
 	 * @param whichNic An InetAddress if binding to a specific network interface. Set to
@@ -139,11 +146,10 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 	protected ServerSocket createServerSocket(int port, int backlog, InetAddress whichNic) throws IOException {
 		ServerSocketFactory serverSocketFactory = this.tcpSocketFactorySupport.getServerSocketFactory();
 		if (whichNic == null) {
-			return serverSocketFactory.createServerSocket(port,
-					Math.abs(backlog));
-		} else {
-			return serverSocketFactory.createServerSocket(port,
-					Math.abs(backlog), whichNic);
+			return serverSocketFactory.createServerSocket(port, Math.abs(backlog));
+		}
+		else {
+			return serverSocketFactory.createServerSocket(port, Math.abs(backlog), whichNic);
 		}
 	}
 
@@ -171,8 +177,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		return tcpSocketFactorySupport;
 	}
 
-	public void setTcpSocketFactorySupport(
-			TcpSocketFactorySupport tcpSocketFactorySupport) {
+	public void setTcpSocketFactorySupport(TcpSocketFactorySupport tcpSocketFactorySupport) {
 		Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null");
 		this.tcpSocketFactorySupport = tcpSocketFactorySupport;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -61,6 +61,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 		super(port);
 	}
 
+	@Override
+	public String getComponentType() {
+		return "tcp-nio-server-connection-factory";
+	}
+
 	/**
 	 * If no listener registers, exits.
 	 * Accepts incoming connections and creates TcpConnections for each new connection.
@@ -70,39 +75,41 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	 */
 	@Override
 	public void run() {
-		if (this.getListener() == null) {
+		if (getListener() == null) {
 			logger.info("No listener bound to server connection factory; will not read; exiting...");
 			return;
 		}
 		try {
 			this.serverChannel = ServerSocketChannel.open();
-			int port = this.getPort();
-			this.getTcpSocketSupport().postProcessServerSocket(this.serverChannel.socket());
+			int port = getPort();
+			getTcpSocketSupport().postProcessServerSocket(this.serverChannel.socket());
 			if (logger.isInfoEnabled()) {
 				logger.info("Listening on port " + port);
 			}
 			this.serverChannel.configureBlocking(false);
-			if (this.getLocalAddress() == null) {
-				this.serverChannel.socket().bind(new InetSocketAddress(port),
-					Math.abs(this.getBacklog()));
+			if (getLocalAddress() == null) {
+				this.serverChannel.socket().bind(new InetSocketAddress(port), Math.abs(getBacklog()));
 			}
 			else {
-				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
-				this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port),
-						Math.abs(this.getBacklog()));
+				InetAddress whichNic = InetAddress.getByName(getLocalAddress());
+				this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port), Math.abs(getBacklog()));
 			}
 			final Selector selector = Selector.open();
 			this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
-			this.setListening(true);
+			setListening(true);
 			this.selector = selector;
 			doSelect(this.serverChannel, selector);
 
 		}
 		catch (IOException e) {
-			this.stop();
+			if (isActive()) {
+				logger.error("Error on ServerChannel; port = " + getPort(), e);
+				publishServerExceptionEvent(e);
+			}
+			stop();
 		}
 		finally {
-			this.setListening(false);
+			setListening(false);
 		}
 	}
 
@@ -119,8 +126,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	 * @throws IOException
 	 */
 	private void doSelect(ServerSocketChannel server, final Selector selector) throws IOException {
-		while (this.isActive()) {
-			int soTimeout = this.getSoTimeout();
+		while (isActive()) {
+			int soTimeout = getSoTimeout();
 			int selectionCount = 0;
 			try {
 				long timeout = soTimeout < 0 ? 0 : soTimeout;
@@ -131,7 +138,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 					logger.trace("Delayed reads:" + getDelayedReads().size() + " timeout " + timeout);
 				}
 				selectionCount = selector.select(timeout);
-				this.processNioSelections(selectionCount, selector, server, this.channelMap);
+				processNioSelections(selectionCount, selector, server, this.channelMap);
 			}
 			catch (CancelledKeyException cke) {
 				if (logger.isDebugEnabled()) {
@@ -139,8 +146,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				}
 			}
 			catch (ClosedSelectorException cse) {
-				if (this.isActive()) {
+				if (isActive()) {
 					logger.error("Selector closed", cse);
+					publishServerExceptionEvent(cse);
 					break;
 				}
 			}
@@ -157,7 +165,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
 		logger.debug("New accept");
 		SocketChannel channel = server.accept();
-		if (this.isShuttingDown()) {
+		if (isShuttingDown()) {
 			if (logger.isInfoEnabled()) {
 				logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress()
 						+ " rejected; the server is in the process of shutting down.");
@@ -173,7 +181,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				if (connection == null) {
 					return;
 				}
-				connection.setTaskExecutor(this.getTaskExecutor());
+				connection.setTaskExecutor(getTaskExecutor());
 				connection.setLastRead(now);
 				this.channelMap.put(channel, connection);
 				channel.register(selector, SelectionKey.OP_READ, connection);
@@ -188,12 +196,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 
 	private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {
 		try {
-			TcpNioConnection connection = this.tcpNioConnectionSupport
-					.createNewConnection(socketChannel, true,
-							this.isLookupHost(), this.getApplicationEventPublisher(), this.getComponentName());
+			TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(socketChannel, true,
+							isLookupHost(), getApplicationEventPublisher(), getComponentName());
 			connection.setUsingDirectBuffers(this.usingDirectBuffers);
 			TcpConnectionSupport wrappedConnection = wrapConnection(connection);
-			this.initializeConnection(wrappedConnection, socketChannel.socket());
+			initializeConnection(wrappedConnection, socketChannel.socket());
 			return connection;
 		}
 		catch (Exception e) {
@@ -204,7 +211,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 
 	@Override
 	public void stop() {
-		this.setActive(false);
+		setActive(false);
 		if (this.selector != null) {
 			try {
 				this.selector.close();

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

@@ -16,25 +16,49 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.endsWith;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
+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.anyString;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 
 import java.io.OutputStream;
+import java.net.BindException;
+import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
+import javax.net.ServerSocketFactory;
+
+import org.apache.commons.logging.Log;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
+import org.mockito.internal.stubbing.answers.DoesNothing;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.serializer.Serializer;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.util.SocketUtils;
 
 /**
  * @author Gary Russell
@@ -44,7 +68,7 @@ import org.springframework.messaging.support.GenericMessage;
 public class ConnectionEventTests {
 
 	@Test
-	public void test() throws Exception {
+	public void testConnectionEvents() throws Exception {
 		Socket socket = mock(Socket.class);
 		final List<TcpConnectionEvent> theEvent = new ArrayList<TcpConnectionEvent>();
 		TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() {
@@ -56,9 +80,9 @@ public class ConnectionEventTests {
 
 			@Override
 			public void publishEvent(Object event) {
-				
+
 			}
-			
+
 		}, "foo");
 		/*
 		 *  Open is not published by the connection itself; the factory publishes it after initialization.
@@ -85,7 +109,72 @@ public class ConnectionEventTests {
 		assertSame(toBeThrown, event.getCause());
 		assertTrue(theEvent.size() > 1);
 		assertNotNull(theEvent.get(1));
-		assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
+		assertTrue(theEvent.get(1).toString()
+				.endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
+	}
+
+	@Test
+	public void testNetServerExceptionEvent() throws Exception {
+		int port = SocketUtils.findAvailableTcpPort();
+		AbstractServerConnectionFactory factory = new TcpNetServerConnectionFactory(port);
+		testServerExceptionGuts(port, factory);
+	}
+
+	@Test
+	public void testNioServerExceptionEvent() throws Exception {
+		int port = SocketUtils.findAvailableTcpPort();
+		AbstractServerConnectionFactory factory = new TcpNioServerConnectionFactory(port);
+		testServerExceptionGuts(port, factory);
+	}
+
+	private void testServerExceptionGuts(int port, AbstractServerConnectionFactory factory) throws Exception {
+		ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(port);
+		final AtomicReference<TcpConnectionServerExceptionEvent> theEvent =
+				new AtomicReference<TcpConnectionServerExceptionEvent>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		factory.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				theEvent.set((TcpConnectionServerExceptionEvent) event);
+				latch.countDown();
+			}
+
+			@Override
+			public void publishEvent(Object event) {
+
+			}
+
+		});
+		factory.setBeanName("sf");
+		factory.registerListener(new TcpListener() {
+
+			@Override
+			public boolean onMessage(Message<?> message) {
+				return false;
+			}
+		});
+		Log logger = spy(TestUtils.getPropertyValue(factory, "logger", Log.class));
+		doAnswer(new DoesNothing()).when(logger).error(anyString(), any(Throwable.class));
+		new DirectFieldAccessor(factory).setPropertyValue("logger", logger);
+
+		factory.start();
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		String actual = theEvent.toString();
+		assertThat(actual, containsString("cause=java.net.BindException"));
+		assertThat(actual, containsString("[factory="
+				+ (factory instanceof TcpNetServerConnectionFactory
+				? "tcp-net-server-connection-factory"
+				: "tcp-nio-server-connection-factory")
+				+ ":sf, port=" + port + "]"));
+
+		ArgumentCaptor<String> reasonCaptor = ArgumentCaptor.forClass(String.class);
+		ArgumentCaptor<Throwable> throwableCaptor = ArgumentCaptor.forClass(Throwable.class);
+		verify(logger).error(reasonCaptor.capture(), throwableCaptor.capture());
+		assertThat(reasonCaptor.getValue(), startsWith("Error on Server"));
+		assertThat(reasonCaptor.getValue(), endsWith("; port = " + port));
+		assertThat(throwableCaptor.getValue(), instanceOf(BindException.class));
+		ss.close();
 	}
 
 }

spring-integration-ip/src/test/resources/log4j.properties

@@ -2,7 +2,6 @@ log4j.rootCategory=WARN, stdout
 
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d %5p %c{1} [%t] : %m%n
 
 log4j.category.org.springframework.integration=WARN
-log4j.category.org.springframework.integration.ip=WARN

src/reference/docbook/ip.xml

@@ -508,6 +508,11 @@
         point the exception occurred. Applications can use a normal <interfacename>ApplicationListener</interfacename>,
         or see <xref linkend="applicationevent-inbound"/>, to capture these events, allowing analysis of the problem.
     </para>
+    <para>
+        Starting with <emphasis>versions 4.0.7, 4.1.3</emphasis> <classname>TcpConnectionServerExceptionEvent</classname>s
+        are published whenever an unexpected exception occurs on a server socket (such as a <classname>BindException</classname>
+        when the server socket is in use). These events have a reference to the connection factory and the cause.
+    </para>
   </section>
   <section id="tcp-adapters">
     <title>TCP Adapters</title>

src/reference/docbook/whats-new.xml

@@ -39,5 +39,13 @@
 				<code>org.springframework.integration.scattergather</code>.
 			</para>
 		</section>
+		<section id="4.2-tcp-server-exceptions">
+			<title>Server Socket Exceptions</title>
+			<para>
+				<classname>TcpConnectionServerExceptionEvent</classname>s are now published whenever an
+				unexpected exception occurs on a TCP server socket (also added to 4.1.3, 4.0.7).
+				See <xref linkend="tcp-events"/> for more information.
+			</para>
+		</section>
 	</section>
 </chapter>