diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index c0d1b6621a..e216c8bc8b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/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); + } + } + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerListeningEvent.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerListeningEvent.java new file mode 100644 index 0000000000..24d1473e50 --- /dev/null +++ b/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; + } + +} 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 57087e00cd..332f2de64e 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 @@ -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; /* 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 1ad999dd92..1c142d7d6a 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 @@ -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); } 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 98d6845fec..10977da1ec 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,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 events = - Collections.synchronizedList(new ArrayList()); + final List events = + Collections.synchronizedList(new ArrayList()); + 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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index 2a18f75bed..1c0a978381 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java index 8c375846ac..faf793a0ca 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java +++ b/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 diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 47720afa79..8e2bd0ace7 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index acd600abd5..ea4ccd0f4e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -38,3 +38,8 @@ See <> 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 <> note for more information. + +==== TCP Changes + +A new `TcpConnectionServerListeningEvent` is emitted when a server connection factory is started. +See <> for more information.