INT-3912: TCP Server Listening Event

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

Polishing - PR Comments

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

@@ -20,9 +20,13 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketAddress;
 import java.net.SocketException;
+import java.util.Date;
 
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.task.TaskRejectedException;
 import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.scheduling.SchedulingAwareRunnable;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 
 /**
@@ -198,4 +202,30 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 		}
 	}
 
+	protected void publishServerListeningEvent(int port) {
+		final ApplicationEventPublisher eventPublisher = getApplicationEventPublisher();
+		if (eventPublisher != null) {
+			final TcpConnectionServerListeningEvent event = new TcpConnectionServerListeningEvent(this, port);
+			TaskScheduler taskScheduler = this.getTaskScheduler();
+			if (taskScheduler != null) {
+				try {
+					taskScheduler.schedule(new Runnable() {
+
+						@Override
+						public void run() {
+							eventPublisher.publishEvent(event);
+						}
+
+					}, new Date());
+				}
+				catch (TaskRejectedException e) {
+					eventPublisher.publishEvent(event);
+				}
+			}
+			else {
+				eventPublisher.publishEvent(event);
+			}
+		}
+	}
+
 }

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

@@ -0,0 +1,46 @@
+/*
+ * 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} emitted when a server begins listening. Useful
+ * when the configured port is zero and the operating system chooses the port.
+ * Also useful to avoid polling the {@code isListening()} if you need to wait
+ * before starting some other process to connect to the socket.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ */
+@SuppressWarnings("serial")
+public class TcpConnectionServerListeningEvent extends IpIntegrationEvent {
+
+	private final int port;
+
+	public TcpConnectionServerListeningEvent(TcpServerConnectionFactory connectionFactory, int port) {
+		super(connectionFactory);
+		Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
+		this.port = port;
+	}
+
+	public int getPort() {
+		return port;
+	}
+
+}

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

@@ -58,8 +58,9 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 	@Override
 	public int getPort() {
 		int port = super.getPort();
-		if (port == 0 && this.serverSocket != null) {
-			return this.serverSocket.getLocalPort();
+		ServerSocket serverSocket = this.serverSocket;
+		if (port == 0 && serverSocket != null) {
+			port = serverSocket.getLocalPort();
 		}
 		return port;
 	}
@@ -100,6 +101,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 			this.serverSocket = theServerSocket;
 			setListening(true);
 			logger.info(this + " Listening");
+			publishServerListeningEvent(getPort());
 			while (true) {
 				final Socket socket;
 				/*

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

@@ -70,9 +70,10 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	@Override
 	public int getPort() {
 		int port = super.getPort();
-		if (port == 0 && this.serverChannel != null) {
+		ServerSocketChannel serverChannel = this.serverChannel;
+		if (port == 0 && serverChannel != null) {
 			try {
-				SocketAddress address = this.serverChannel.getLocalAddress();
+				SocketAddress address = serverChannel.getLocalAddress();
 				if (address instanceof InetSocketAddress) {
 					port = ((InetSocketAddress) address).getPort();
 				}
@@ -132,6 +133,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 			else {
 				this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
 				setListening(true);
+				publishServerListeningEvent(getPort());
 				this.selector = selector;
 				doSelect(this.serverChannel, selector);
 			}

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

@@ -16,8 +16,10 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
@@ -47,12 +49,16 @@ import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.ip.event.IpIntegrationEvent;
 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;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 /**
  * @author Gary Russell
@@ -75,13 +81,17 @@ public class ConnectionFactoryTests {
 	}
 
 	public void testObtainConnectionIds(AbstractServerConnectionFactory serverFactory) throws Exception {
-		final List<TcpConnectionEvent> events =
-				Collections.synchronizedList(new ArrayList<TcpConnectionEvent>());
+		final List<IpIntegrationEvent> events =
+				Collections.synchronizedList(new ArrayList<IpIntegrationEvent>());
+		final CountDownLatch serverListeningLatch = new CountDownLatch(1);
 		ApplicationEventPublisher publisher = new ApplicationEventPublisher() {
 
 			@Override
 			public void publishEvent(ApplicationEvent event) {
-				events.add((TcpConnectionEvent) event);
+				events.add((IpIntegrationEvent) event);
+				if (event instanceof TcpConnectionServerListeningEvent) {
+					serverListeningLatch.countDown();
+				}
 			}
 
 			@Override
@@ -102,10 +112,19 @@ public class ConnectionFactoryTests {
 				return result;
 			}
 		}).when(serverFactory).wrapConnection(any(TcpConnectionSupport.class));
+		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+		scheduler.setPoolSize(10);
+		scheduler.afterPropertiesSet();
+		BeanFactory bf = mock(BeanFactory.class);
+		when(bf.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
+		when(bf.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)).thenReturn(scheduler);
+		serverFactory.setBeanFactory(bf);
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(serverFactory);
 		adapter.start();
-		TestingUtilities.waitListening(serverFactory, null);
+		assertTrue("Listening event not received", serverListeningLatch.await(10, TimeUnit.SECONDS));
+		assertThat(events.get(0), instanceOf(TcpConnectionServerListeningEvent.class));
+		assertThat(((TcpConnectionServerListeningEvent) events.get(0)).getPort(), equalTo(serverFactory.getPort()));
 		int port = serverFactory.getPort();
 		TcpNetClientConnectionFactory clientFactory = new TcpNetClientConnectionFactory("localhost", port);
 		clientFactory.registerListener(new TcpListener() {
@@ -127,11 +146,15 @@ public class ConnectionFactoryTests {
 		assertTrue(serverFactory.closeConnection(servers.get(0)));
 		servers = serverFactory.getOpenConnectionIds();
 		assertEquals(0, servers.size());
-		Thread.sleep(1000);
+		int n = 0;
 		clients = clientFactory.getOpenConnectionIds();
+		while (n++ < 100 && clients.size() > 0) {
+			Thread.sleep(100);
+			clients = clientFactory.getOpenConnectionIds();
+		}
 		assertEquals(0, clients.size());
-		int expected = serverFactory instanceof TcpNetServerConnectionFactory ? 6// OPEN, CLOSE, EXCEPTION for each side
-				: 4; //OPEN, CLOSE (but we *might* get exceptions, depending on timing).
+		int expected = serverFactory instanceof TcpNetServerConnectionFactory ? 7// Listening, OPEN, CLOSE, EXCEPTION for each side
+				: 5; //Listening, OPEN, CLOSE (but we *might* get exceptions, depending on timing).
 		assertThat(events.size(), greaterThanOrEqualTo(expected));
 
 		FooEvent event = new FooEvent(client, "foo");
@@ -153,6 +176,7 @@ public class ConnectionFactoryTests {
 			assertEquals(port, inetAddress.getPort());
 		}
 		serverFactory.stop();
+		scheduler.shutdown();
 	}
 
 	@Test

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

@@ -728,7 +728,9 @@ public class TcpNioConnectionTests {
 
 			@Override
 			public void publishEvent(ApplicationEvent event) {
-				connectionLatch.countDown();
+				if (event instanceof TcpConnectionOpenEvent) {
+					connectionLatch.countDown();
+				}
 			}
 
 			@Override

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java

@@ -82,7 +82,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
 		test(location);
 	}
 
-	private void test(String location) throws InterruptedException, Exception {
+	private void test(String location) throws Exception {
 		UdpMulticastEndToEndTests launcher = new UdpMulticastEndToEndTests();
 		Thread t = new Thread(launcher);
 		t.start(); // launch the receiver

src/reference/asciidoc/ip.adoc

@@ -405,6 +405,14 @@ collaborating outbound channel adapter) receives a message that cannot be routed
 Outbound gateways also publish this event when a late reply is received (the sender thread has timed out).
 The event contains the connection id as well as an exception in the `cause` property that contains the failed message.
 
+Starting with _version 4.3_, a `TcpConnectionServerListeningEvent` is emitted when a server connection factory is started.
+This is useful when the factory is configured to listen on port 0, meaning that the operating system chooses the port.
+It can also be used instead of polling `isListening()`, if you need to wait before starting some other process that will
+connect to the socket.
+
+IMPORTANT: To avoid delaying the listening thread from accepting connections, the event is published on a separate
+thread.
+
 [[tcp-adapters]]
 === TCP Adapters
 

src/reference/asciidoc/whats-new.adoc

@@ -38,3 +38,8 @@ See <<jms-header-mapping>> for more information.
 There is a change in behavior when a POJO aggregator releases a collection of `Message<?>` objects; this is rare but if
 your application does that, you will need to make a small change to your POJO. See this <<agg-message-collection>> note
 for more information.
+
+==== TCP Changes
+
+A new `TcpConnectionServerListeningEvent` is emitted when a server connection factory is started.
+See <<tcp-events>> for more information.