INT-3984: UDP: OS Selected Server Port

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

Allow UDP server port (and sender Ack port) to be selected by the operating system.

Also fix multicast tests to run on all platforms.

Change tests to use OS selected port.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 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.
@@ -17,8 +17,8 @@
 package org.springframework.integration.ip.udp;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.net.DatagramPacket;
@@ -26,10 +26,11 @@ import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 
 import org.junit.Test;
-import org.springframework.messaging.Message;
+
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.mapping.MessageMappingException;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
 
 /**
  * @author Gary Russell
@@ -69,7 +70,7 @@ public class DatagramPacketMessageMapperTests {
 		Message<byte[]> messageOut = mapper.toMessage(packet);
 		assertEquals(new String(message.getPayload()), new String(messageOut.getPayload()));
 		if (ack) {
-			assertEquals(messageOut.getHeaders().get(IpHeaders.ACK_ID).toString(), 
+			assertEquals(messageOut.getHeaders().get(IpHeaders.ACK_ID).toString(),
 					     message.getHeaders().getId().toString());
 		}
 		assertTrue(((String)messageOut.getHeaders().get(IpHeaders.HOSTNAME)).contains("localhost"));
@@ -77,10 +78,10 @@ public class DatagramPacketMessageMapperTests {
 		messageOut = mapper.toMessage(packet);
 		assertEquals(new String(message.getPayload()), new String(messageOut.getPayload()));
 		if (ack) {
-			assertEquals(messageOut.getHeaders().get(IpHeaders.ACK_ID).toString(), 
+			assertEquals(messageOut.getHeaders().get(IpHeaders.ACK_ID).toString(),
 					     message.getHeaders().getId().toString());
 		}
-		assertFalse(((String)messageOut.getHeaders().get(IpHeaders.HOSTNAME)).contains("localhost"));		
+		assertFalse(((String)messageOut.getHeaders().get(IpHeaders.HOSTNAME)).contains("localhost"));
 	}
 
 	@Test

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.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.
@@ -25,20 +25,18 @@ import java.net.DatagramSocket;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.MulticastSocket;
-import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.LogFactory;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.support.MessageBuilder;
-import org.springframework.integration.test.util.SocketUtils;
 import org.springframework.messaging.Message;
 
 /**
@@ -52,17 +50,20 @@ public class DatagramPacketSendingHandlerTests {
 
 	@Test
 	public void verifySend() throws Exception {
-		final int testPort = SocketUtils.findAvailableUdpSocket();
 		byte[] buffer = new byte[8];
 		final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
-		final CountDownLatch latch = new CountDownLatch(1);
+		final CountDownLatch received = new CountDownLatch(1);
+		final AtomicInteger testPort = new AtomicInteger();
+		final CountDownLatch listening = new CountDownLatch(1);
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
 			@Override
 			public void run() {
 				try {
-					DatagramSocket socket = new DatagramSocket(testPort);
+					DatagramSocket socket = new DatagramSocket();
+					testPort.set(socket.getLocalPort());
+					listening.countDown();
 					socket.receive(receivedPacket);
-					latch.countDown();
+					received.countDown();
 					socket.close();
 				}
 				catch (Exception e) {
@@ -70,12 +71,12 @@ public class DatagramPacketSendingHandlerTests {
 				}
 			}
 		});
-		Thread.sleep(1000);
+		assertTrue(listening.await(10, TimeUnit.SECONDS));
 		UnicastSendingMessageHandler handler =
-				new UnicastSendingMessageHandler("localhost", testPort);
+				new UnicastSendingMessageHandler("localhost", testPort.get());
 		String payload = "foo";
 		handler.handleMessage(MessageBuilder.withPayload(payload).build());
-		assertTrue(latch.await(3000, TimeUnit.MILLISECONDS));
+		assertTrue(received.await(3000, TimeUnit.MILLISECONDS));
 		byte[] src = receivedPacket.getData();
 		int length = receivedPacket.getLength();
 		int offset = receivedPacket.getOffset();
@@ -88,27 +89,22 @@ public class DatagramPacketSendingHandlerTests {
 	@Test
 	public void verifySendWithAck() throws Exception {
 
-		final List<Integer> openPorts = SocketUtils.findAvailableUdpSockets(SocketUtils.getRandomSeedPort(), 2);
-
-		final int testPort = openPorts.get(0);
-		final int ackPort = openPorts.get(1);
+		final AtomicInteger testPort = new AtomicInteger();
+		final AtomicInteger ackPort = new AtomicInteger();
 
 		byte[] buffer = new byte[1000];
 		final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
-		final CountDownLatch latch1 = new CountDownLatch(1);
-		final CountDownLatch latch2 = new CountDownLatch(1);
-		UnicastSendingMessageHandler handler =
-				new UnicastSendingMessageHandler("localhost", testPort, true,
-						true, "localhost", ackPort, 5000);
-		handler.setBeanFactory(mock(BeanFactory.class));
-		handler.afterPropertiesSet();
-		handler.start();
+		final CountDownLatch listening = new CountDownLatch(1);
+		final CountDownLatch ackListening = new CountDownLatch(1);
+		final CountDownLatch ackSent = new CountDownLatch(1);
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
 			@Override
 			public void run() {
 				try {
-					DatagramSocket socket = new DatagramSocket(testPort);
-					latch1.countDown();
+					DatagramSocket socket = new DatagramSocket();
+					testPort.set(socket.getLocalPort());
+					listening.countDown();
+					assertTrue(ackListening.await(10, TimeUnit.SECONDS));
 					socket.receive(receivedPacket);
 					socket.close();
 					DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
@@ -118,21 +114,29 @@ public class DatagramPacketSendingHandlerTests {
 					Object id = message.getHeaders().get(IpHeaders.ACK_ID);
 					byte[] ack = id.toString().getBytes();
 					DatagramPacket ackPack = new DatagramPacket(ack, ack.length,
-							                        new InetSocketAddress("localHost", ackPort));
+							                        new InetSocketAddress("localHost", ackPort.get()));
 					DatagramSocket out = new DatagramSocket();
 					out.send(ackPack);
 					out.close();
-					latch2.countDown();
+					ackSent.countDown();
 				}
 				catch (Exception e) {
 					e.printStackTrace();
 				}
 			}
 		});
-		latch1.await(3000, TimeUnit.MILLISECONDS);
+		listening.await(10000, TimeUnit.MILLISECONDS);
+		UnicastSendingMessageHandler handler =
+				new UnicastSendingMessageHandler("localhost", testPort.get(), true, true, "localhost", 0, 5000);
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.afterPropertiesSet();
+		handler.start();
+		waitAckListening(handler);
+		ackPort.set(handler.getAckPort());
+		ackListening.countDown();
 		String payload = "foobar";
 		handler.handleMessage(MessageBuilder.withPayload(payload).build());
-		assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS));
+		assertTrue(ackSent.await(10000, TimeUnit.MILLISECONDS));
 		byte[] src = receivedPacket.getData();
 		int length = receivedPacket.getLength();
 		int offset = receivedPacket.getOffset();
@@ -142,14 +146,28 @@ public class DatagramPacketSendingHandlerTests {
 		handler.stop();
 	}
 
+	public void waitAckListening(UnicastSendingMessageHandler handler) throws InterruptedException {
+		int n = 0;
+		while (n++ < 100 && handler.getAckPort() == 0) {
+			Thread.sleep(100);
+		}
+		assertTrue(n < 100);
+	}
+
 	@Test
-	@Ignore
 	public void verifySendMulticast() throws Exception {
-		final int testPort = SocketUtils.findAvailableUdpSocket();
+		MulticastSocket socket;
+		try {
+			socket = new MulticastSocket();
+		}
+		catch (Exception e) {
+			return;
+		}
+		final int testPort = socket.getLocalPort();
 		final String multicastAddress = "225.6.7.8";
 		final String payload = "foo";
-		final CountDownLatch latch1 = new CountDownLatch(2);
-		final CountDownLatch latch2 = new CountDownLatch(2);
+		final CountDownLatch listening = new CountDownLatch(2);
+		final CountDownLatch received = new CountDownLatch(2);
 		Runnable catcher = new Runnable() {
 			@Override
 			public void run() {
@@ -159,7 +177,7 @@ public class DatagramPacketSendingHandlerTests {
 					MulticastSocket socket = new MulticastSocket(testPort);
 					InetAddress group = InetAddress.getByName(multicastAddress);
 					socket.joinGroup(group);
-					latch1.countDown();
+					listening.countDown();
 					LogFactory.getLog(getClass())
 						.debug(Thread.currentThread().getName() + " waiting for packet");
 					socket.receive(receivedPacket);
@@ -172,11 +190,11 @@ public class DatagramPacketSendingHandlerTests {
 					assertEquals(payload, new String(dest));
 					LogFactory.getLog(getClass())
 						.debug(Thread.currentThread().getName() + " received packet");
-					latch2.countDown();
+					received.countDown();
 				}
 				catch (Exception e) {
 					noMulticast = true;
-					latch1.countDown();
+					listening.countDown();
 					e.printStackTrace();
 				}
 			}
@@ -184,29 +202,36 @@ public class DatagramPacketSendingHandlerTests {
 		Executor executor = Executors.newFixedThreadPool(2);
 		executor.execute(catcher);
 		executor.execute(catcher);
-		latch1.await(3000, TimeUnit.MILLISECONDS);
+		listening.await(10000, TimeUnit.MILLISECONDS);
 		if (noMulticast) {
+			socket.close();
 			return;
 		}
 		MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort);
 		handler.handleMessage(MessageBuilder.withPayload(payload).build());
-		assertTrue(latch2.await(3000, TimeUnit.MILLISECONDS));
+		assertTrue(received.await(10000, TimeUnit.MILLISECONDS));
 		handler.stop();
+		socket.close();
 	}
 
 	@Test
-	@Ignore
 	public void verifySendMulticastWithAcks() throws Exception {
 
-		final List<Integer> openPorts = SocketUtils.findAvailableUdpSockets(SocketUtils.getRandomSeedPort(), 2);
-
-		final int testPort = openPorts.get(0);
-		final int ackPort = openPorts.get(1);
+		MulticastSocket socket;
+		try {
+			socket = new MulticastSocket();
+		}
+		catch (Exception e) {
+			return;
+		}
+		final int testPort = socket.getLocalPort();
+		final AtomicInteger ackPort = new AtomicInteger();
 
 		final String multicastAddress = "225.6.7.8";
 		final String payload = "foobar";
-		final CountDownLatch latch1 = new CountDownLatch(2);
-		final CountDownLatch latch2 = new CountDownLatch(2);
+		final CountDownLatch listening = new CountDownLatch(2);
+		final CountDownLatch ackListening = new CountDownLatch(1);
+		final CountDownLatch ackSent = new CountDownLatch(2);
 		Runnable catcher = new Runnable() {
 			@Override
 			public void run() {
@@ -214,9 +239,11 @@ public class DatagramPacketSendingHandlerTests {
 					byte[] buffer = new byte[1000];
 					DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
 					MulticastSocket socket = new MulticastSocket(testPort);
+					socket.setSoTimeout(8000);
 					InetAddress group = InetAddress.getByName(multicastAddress);
 					socket.joinGroup(group);
-					latch1.countDown();
+					listening.countDown();
+					assertTrue(ackListening.await(10, TimeUnit.SECONDS));
 					LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " waiting for packet");
 					socket.receive(receivedPacket);
 					socket.close();
@@ -234,15 +261,16 @@ public class DatagramPacketSendingHandlerTests {
 					Object id = message.getHeaders().get(IpHeaders.ACK_ID);
 					byte[] ack = id.toString().getBytes();
 					DatagramPacket ackPack = new DatagramPacket(ack, ack.length,
-							                        new InetSocketAddress("localHost", ackPort));
+												new InetSocketAddress("localHost", ackPort.get()));
 					DatagramSocket out = new DatagramSocket();
 					out.send(ackPack);
 					out.close();
-					latch2.countDown();
+					ackSent.countDown();
+					socket.close();
 				}
 				catch (Exception e) {
 					noMulticast = true;
-					latch1.countDown();
+					listening.countDown();
 					e.printStackTrace();
 				}
 			}
@@ -250,19 +278,23 @@ public class DatagramPacketSendingHandlerTests {
 		Executor executor = Executors.newFixedThreadPool(2);
 		executor.execute(catcher);
 		executor.execute(catcher);
-		latch1.await(3000, TimeUnit.MILLISECONDS);
-		if (noMulticast) {
+		listening.await(10000, TimeUnit.MILLISECONDS);
+		if (this.noMulticast) {
+			socket.close();
 			return;
 		}
 		MulticastSendingMessageHandler handler =
-			new MulticastSendingMessageHandler(multicastAddress, testPort, true,
-                    							true, "localhost", ackPort, 500000);
+			new MulticastSendingMessageHandler(multicastAddress, testPort, true, true, "localhost", 0, 10000);
 		handler.setMinAcksForSuccess(2);
 		handler.afterPropertiesSet();
 		handler.start();
+		waitAckListening(handler);
+		ackPort.set(handler.getAckPort());
+		ackListening.countDown();
 		handler.handleMessage(MessageBuilder.withPayload(payload).build());
-		assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS));
+		assertTrue(ackSent.await(10000, TimeUnit.MILLISECONDS));
 		handler.stop();
+		socket.close();
 	}
 
 }

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

@@ -26,7 +26,9 @@ import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.Inet4Address;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.MulticastSocket;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
@@ -36,7 +38,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.LogFactory;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
@@ -61,22 +62,26 @@ public class UdpChannelAdapterTests {
 
 	@Test
 	public void testUnicastReceiver() throws Exception {
-		testUnicastReceiver(false);
+		testUnicastReceiver(false, false);
+	}
+
+	@Test
+	public void testUnicastReceiverLocalNicOnly() throws Exception {
+		testUnicastReceiver(false, true);
 	}
 
 	@Test
 	public void testUnicastReceiverDeadExecutor() throws Exception {
-		testUnicastReceiver(true);
+		testUnicastReceiver(true, false);
 	}
 
-	private void testUnicastReceiver(final boolean killExecutor) throws Exception {
+	private void testUnicastReceiver(final boolean killExecutor, boolean useLocalAddress) throws Exception {
 		QueueChannel channel = new QueueChannel(2);
-		int port = SocketUtils.findAvailableUdpSocket();
 		final CountDownLatch stopLatch = new CountDownLatch(1);
 		final CountDownLatch exitLatch = new CountDownLatch(1);
 		final AtomicBoolean stopping = new AtomicBoolean();
 		final AtomicReference<Exception> exceptionHolder = new AtomicReference<Exception>();
-		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port) {
+		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(0) {
 
 			@Override
 			public boolean isActive() {
@@ -130,9 +135,12 @@ public class UdpChannelAdapterTests {
 
 		};
 		adapter.setOutputChannel(channel);
-//		SocketUtils.setLocalNicIfPossible(adapter);
+		if (useLocalAddress) {
+			adapter.setLocalAddress("127.0.0.1");
+		}
 		adapter.start();
 		SocketTestUtils.waitListening(adapter);
+		int port = adapter.getPort();
 
 		Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
 		DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
@@ -157,11 +165,11 @@ public class UdpChannelAdapterTests {
 	@Test
 	public void testUnicastReceiverWithReply() throws Exception {
 		QueueChannel channel = new QueueChannel(2);
-		int port = SocketUtils.findAvailableUdpSocket();
-		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
+		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(0);
 		adapter.setOutputChannel(channel);
 		adapter.start();
 		SocketTestUtils.waitListening(adapter);
+		int port = adapter.getPort();
 
 		Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
 		DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
@@ -211,13 +219,13 @@ public class UdpChannelAdapterTests {
 	@Test
 	public void testUnicastSender() throws Exception {
 		QueueChannel channel = new QueueChannel(2);
-		int port = SocketUtils.findAvailableUdpSocket();
-		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
+		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(0);
 		adapter.setBeanName("test");
 		adapter.setOutputChannel(channel);
 //		SocketUtils.setLocalNicIfPossible(adapter);
 		adapter.start();
 		SocketTestUtils.waitListening(adapter);
+		int port = adapter.getPort();
 
 //		String whichNic = SocketUtils.chooseANic(false);
 		UnicastSendingMessageHandler handler = new UnicastSendingMessageHandler(
@@ -238,20 +246,20 @@ public class UdpChannelAdapterTests {
 	}
 
 	@SuppressWarnings("unchecked")
-	@Test @Ignore
+	@Test
 	public void testMulticastReceiver() throws Exception {
+		System.setProperty("java.net.preferIPv4Stack", "true");
 		QueueChannel channel = new QueueChannel(2);
-		int port = SocketUtils.findAvailableUdpSocket();
-		MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.8", port);
+		MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.8", 0);
 		adapter.setOutputChannel(channel);
-		String nic = SocketTestUtils.chooseANic(true);
-		if (nic == null) {	// no multicast support
-			LogFactory.getLog(this.getClass()).error("No Multicast support");
+		String nic = checkMulticast();
+		if (nic == null) {
 			return;
 		}
 		adapter.setLocalAddress(nic);
 		adapter.start();
 		SocketTestUtils.waitListening(adapter);
+		int port = adapter.getPort();
 
 		Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
 		DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
@@ -267,16 +275,33 @@ public class UdpChannelAdapterTests {
 		adapter.stop();
 	}
 
+	private String checkMulticast() throws Exception {
+		String nic = SocketTestUtils.chooseANic(true);
+		if (nic == null) {	// no multicast support
+			LogFactory.getLog(this.getClass()).info("No Multicast support; test skipped");
+			return null;
+		}
+		try {
+			MulticastSocket socket = new MulticastSocket();
+			socket.joinGroup(InetAddress.getByName("225.6.7.9"));
+			socket.close();
+		}
+		catch (Exception e) {
+			LogFactory.getLog(this.getClass()).info("No Multicast support; test skipped");
+		}
+		return nic;
+	}
+
 	@SuppressWarnings("unchecked")
-	@Test @Ignore
+	@Test
 	public void testMulticastSender() throws Exception {
+		System.setProperty("java.net.preferIPv4Stack", "true");
 		QueueChannel channel = new QueueChannel(2);
 		int port = SocketUtils.findAvailableUdpSocket();
 		UnicastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.9", port);
 		adapter.setOutputChannel(channel);
-		String nic = SocketTestUtils.chooseANic(true);
-		if (nic == null) {	// no multicast support
-			LogFactory.getLog(this.getClass()).error("No Multicast support");
+		String nic = checkMulticast();
+		if (nic == null) {
 			return;
 		}
 		adapter.setLocalAddress(nic);
@@ -297,8 +322,7 @@ public class UdpChannelAdapterTests {
 	@Test
 	public void testUnicastReceiverException() throws Exception {
 		SubscribableChannel channel = new DirectChannel();
-		int port = SocketUtils.findAvailableUdpSocket();
-		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
+		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(0);
 		adapter.setOutputChannel(channel);
 //		SocketUtils.setLocalNicIfPossible(adapter);
 		adapter.setOutputChannel(channel);
@@ -308,6 +332,7 @@ public class UdpChannelAdapterTests {
 		adapter.setErrorChannel(errorChannel);
 		adapter.start();
 		SocketTestUtils.waitListening(adapter);
+		int port = adapter.getPort();
 
 		Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
 		DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();

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

@@ -21,15 +21,17 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.util.Date;
+import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.core.env.PropertiesPropertySource;
+import org.springframework.core.env.StandardEnvironment;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.messaging.Message;
@@ -66,22 +68,44 @@ public class UdpMulticastEndToEndTests implements Runnable {
 
 	private final CountDownLatch readyToReceive = new CountDownLatch(1);
 
+	private volatile int receiverPort;
+
 	private static long hangAroundFor = 0;
 
 
 	@Test
-	@Ignore
 	public void runIt() throws Exception {
+		String location = "org/springframework/integration/ip/udp/testIp-out-multicast-context.xml";
+		test(location);
+	}
+
+	private void test(String location) throws InterruptedException, Exception {
 		UdpMulticastEndToEndTests launcher = new UdpMulticastEndToEndTests();
 		Thread t = new Thread(launcher);
 		t.start(); // launch the receiver
-		AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
-				"testIp-out-multicast-context.xml",
-				UdpMulticastEndToEndTests.class);
+		int n = 0;
+		while (n++ < 100 && launcher.getReceiverPort() == 0) {
+			Thread.sleep(100);
+		}
+		assertTrue("Receiver failed to listen", n < 100);
+
+		ClassPathXmlApplicationContext applicationContext = createContext(launcher, location);
 		launcher.launchSender(applicationContext);
-		applicationContext.stop();
+		applicationContext.close();
 	}
 
+	private ClassPathXmlApplicationContext createContext(UdpMulticastEndToEndTests launcher, String location) {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setConfigLocation(location);
+		StandardEnvironment env = new StandardEnvironment();
+		Properties props = new Properties();
+		props.setProperty("port", Integer.toString(launcher.getReceiverPort()));
+		PropertiesPropertySource pps = new PropertiesPropertySource("ftpprops", props);
+		env.getPropertySources().addLast(pps);
+		applicationContext.setEnvironment(env);
+		applicationContext.refresh();
+		return applicationContext;
+	}
 
 	public void launchSender(ApplicationContext applicationContext) throws Exception {
 		DestinationResolver<MessageChannel> channelResolver = new BeanFactoryChannelResolver(applicationContext);
@@ -118,6 +142,9 @@ public class UdpMulticastEndToEndTests implements Runnable {
 		assertEquals(testingIpText, new String(finalMessage.getPayload()));
 	}
 
+	public int getReceiverPort() {
+		return receiverPort;
+	}
 
 	/**
 	 * Instantiate the receiving context
@@ -128,6 +155,24 @@ public class UdpMulticastEndToEndTests implements Runnable {
 		AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
 				"testIp-in-multicast-context.xml",
 				UdpMulticastEndToEndTests.class);
+		MulticastReceivingChannelAdapter inbound = ctx.getBean(MulticastReceivingChannelAdapter.class);
+		int n = 0;
+		try {
+			while (!inbound.isListening()) {
+				Thread.sleep(100);
+				if (n++ > 100) {
+					throw new RuntimeException("Failed to start listening");
+				}
+			}
+		}
+		catch (RuntimeException e) {
+			throw e;
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new RuntimeException("interrupted");
+		}
+		this.receiverPort = inbound.getPort();
 		while (okToRun) {
 			try {
 				readyToReceive.countDown();

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

@@ -32,6 +32,8 @@ import org.junit.Test;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.core.env.PropertiesPropertySource;
+import org.springframework.core.env.StandardEnvironment;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
@@ -71,6 +73,8 @@ public class UdpUnicastEndToEndTests implements Runnable {
 
 	private CountDownLatch readyToReceive = new CountDownLatch(1);
 
+	private volatile int receiverPort;
+
 	private static long hangAroundFor = 10;
 
 	@Before
@@ -86,24 +90,42 @@ public class UdpUnicastEndToEndTests implements Runnable {
 
 	@Test
 	public void runIt() throws Exception {
-		UdpUnicastEndToEndTests launcher = new UdpUnicastEndToEndTests();
-		Thread t = new Thread(launcher);
-		t.start(); // launch the receiver
-		AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
-				"testIp-out-context.xml", UdpUnicastEndToEndTests.class);
-		launcher.launchSender(applicationContext);
-		applicationContext.stop();
+		String location = "org/springframework/integration/ip/udp/testIp-out-context.xml";
+		test(location);
 	}
 
 	@Test
 	public void tesUdpOutboundChannelAdapterWithinChain() throws Exception {
+		String location = "org/springframework/integration/ip/udp/testIp-out-within-chain-context.xml";
+		test(location);
+	}
+
+	private void test(String location) throws InterruptedException, Exception {
 		UdpUnicastEndToEndTests launcher = new UdpUnicastEndToEndTests();
 		Thread t = new Thread(launcher);
 		t.start(); // launch the receiver
-		AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
-				"testIp-out-within-chain-context.xml", UdpUnicastEndToEndTests.class);
+		int n = 0;
+		while (n++ < 100 && launcher.getReceiverPort() == 0) {
+			Thread.sleep(100);
+		}
+		assertTrue("Receiver failed to listen", n < 100);
+
+		ClassPathXmlApplicationContext applicationContext = createContext(launcher, location);
 		launcher.launchSender(applicationContext);
-		applicationContext.stop();
+		applicationContext.close();
+	}
+
+	private ClassPathXmlApplicationContext createContext(UdpUnicastEndToEndTests launcher, String location) {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setConfigLocation(location);
+		StandardEnvironment env = new StandardEnvironment();
+		Properties props = new Properties();
+		props.setProperty("port", Integer.toString(launcher.getReceiverPort()));
+		PropertiesPropertySource pps = new PropertiesPropertySource("ftpprops", props);
+		env.getPropertySources().addLast(pps);
+		applicationContext.setEnvironment(env);
+		applicationContext.refresh();
+		return applicationContext;
 	}
 
 
@@ -138,6 +160,9 @@ public class UdpUnicastEndToEndTests implements Runnable {
 		assertEquals(testingIpText, new String(finalMessage.getPayload()));
 	}
 
+	public int getReceiverPort() {
+		return receiverPort;
+	}
 
 	/**
 	 * Instantiate the receiving context
@@ -158,7 +183,14 @@ public class UdpUnicastEndToEndTests implements Runnable {
 				}
 			}
 		}
-		catch (Exception e) { }
+		catch (RuntimeException e) {
+			throw e;
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new RuntimeException("interrupted");
+		}
+		this.receiverPort = inbound.getPort();
 		while (okToRun) {
 			try {
 				readyToReceive.countDown();

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml

@@ -12,19 +12,19 @@
 			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
 			http://www.springframework.org/schema/integration/ip
 			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
-	
+
 	<message-history/>
-	<!--  
-		Play with the buffer size to force errors. If the checkLength property is 
+	<!--
+		Play with the buffer size to force errors. If the checkLength property is
 		set and this buffer is too small, we'll throw an exception.
 	-->
 	<ip:udp-inbound-channel-adapter id="udpReceiver"
 		channel="udpOutChannel"
-		port="11111"
+		port="0"
 		receive-buffer-size="500"
 		multicast="false"
 		check-length="true" />
 
 	<beans:import resource="testIp-common-context.xml" />
-	
+
 </beans:beans>

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-multicast-context.xml

@@ -13,18 +13,18 @@
 			http://www.springframework.org/schema/integration/ip
 			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
 
-	<!--  
-		Play with the buffer size to force errors. If the checkLength property is 
+	<!--
+		Play with the buffer size to force errors. If the checkLength property is
 		set and this buffer is too small, we'll throw an exception.
 	-->
 	<ip:udp-inbound-channel-adapter id="mcUdpReceiver"
 		channel="udpOutChannel"
-		port="11112"
+		port="0"
 		receive-buffer-size="500"
 		multicast="true"
 		multicast-address="225.6.7.8"
 		check-length="true" />
 
 	<beans:import resource="testIp-common-context.xml" />
-	
+
 </beans:beans>

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml

@@ -4,14 +4,14 @@
 	xmlns:beans="http://www.springframework.org/schema/beans"
 	xmlns:stream="http://www.springframework.org/schema/integration/stream"
 	xmlns:ip="http://www.springframework.org/schema/integration/ip"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
-			http://www.springframework.org/schema/beans/spring-beans.xsd
-			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd
-			http://www.springframework.org/schema/integration/stream
-			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
-			http://www.springframework.org/schema/integration/ip
-			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
+		http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<context:property-placeholder />
 
 	<stream:stdin-channel-adapter id="stdin" channel="outputChannel" >
 		<poller fixed-delay="100"/>
@@ -26,13 +26,13 @@
 	                   ref="testIp"
 	                   method="testIp"/>
 
-	<ip:udp-outbound-channel-adapter id="udpSender" 
+	<ip:udp-outbound-channel-adapter id="udpSender"
 		host="localhost"
-		port="11111"
+		port="${port}"
 		check-length="true"
 		acknowledge="true"
 		ack-host="localhost"
-		ack-port="22222"
+		ack-port="0"
 		ack-timeout="10000"
 		channel="outputChannel"/>
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml

@@ -4,14 +4,14 @@
 	xmlns:beans="http://www.springframework.org/schema/beans"
 	xmlns:stream="http://www.springframework.org/schema/integration/stream"
 	xmlns:ip="http://www.springframework.org/schema/integration/ip"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
-			http://www.springframework.org/schema/beans/spring-beans.xsd
-			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd
-			http://www.springframework.org/schema/integration/stream
-			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
-			http://www.springframework.org/schema/integration/ip
-			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
+		http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<context:property-placeholder />
 
 	<stream:stdin-channel-adapter id="stdin" channel="mcOutputChannel" >
 		<poller fixed-delay="100" />
@@ -26,18 +26,18 @@
 	                   ref="testIp"
 	                   method="testIp"/>
 
-	<ip:udp-outbound-channel-adapter id="mcUdpSender" 
+	<ip:udp-outbound-channel-adapter id="mcUdpSender"
 		multicast="true"
 		time-to-live="2"
 		host="225.6.7.8"
-		port="11112"
+		port="${port}"
 		check-length="true"
 		acknowledge="true"
 		ack-host="localhost"
-		ack-port="22223"
+		ack-port="0"
 		ack-timeout="10000"
 		channel="mcOutputChannel"/>
 
 	<beans:import resource="testIp-common-context.xml" />
-	 		
+
 </beans:beans>

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-within-chain-context.xml

@@ -4,14 +4,14 @@
 	xmlns:beans="http://www.springframework.org/schema/beans"
 	xmlns:stream="http://www.springframework.org/schema/integration/stream"
 	xmlns:ip="http://www.springframework.org/schema/integration/ip"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
-			http://www.springframework.org/schema/beans/spring-beans.xsd
-			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd
-			http://www.springframework.org/schema/integration/stream
-			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
-			http://www.springframework.org/schema/integration/ip
-			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
+		http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<context:property-placeholder />
 
 	<stream:stdin-channel-adapter id="stdin" channel="outputChannel" >
 		<poller fixed-delay="100"/>
@@ -28,11 +28,11 @@
 
 	<chain input-channel="outputChannel">
 		<ip:udp-outbound-channel-adapter host="localhost"
-										 port="11111"
+										 port="${port}"
 										 check-length="true"
 										 acknowledge="true"
 										 ack-host="localhost"
-										 ack-port="22223"
+										 ack-port="0"
 										 ack-timeout="10000"/>
 	</chain>
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java

@@ -19,10 +19,13 @@ package org.springframework.integration.ip.util;
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.io.OutputStream;
+import java.net.Inet4Address;
 import java.net.InetAddress;
+import java.net.NetworkInterface;
 import java.net.Socket;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
+import java.util.Enumeration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -431,17 +434,19 @@ public class SocketTestUtils {
 	}
 
 	public static String chooseANic(boolean multicast) throws Exception {
-//		Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
-//		while (interfaces.hasMoreElements()) {
-//			NetworkInterface intface = interfaces.nextElement();
-//			if (intface.isLoopback() || (multicast && !intface.supportsMulticast()))
-//				continue;
-//			Enumeration<InetAddress> inet = intface.getInetAddresses();
-//			if (!inet.hasMoreElements())
-//				continue;
-//			String address = inet.nextElement().getHostAddress();
-//			return address;
-//		}
+		Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
+		while (interfaces.hasMoreElements()) {
+			NetworkInterface intface = interfaces.nextElement();
+			if (intface.isLoopback() || (multicast && !intface.supportsMulticast())) {
+				continue;
+			}
+			for (Enumeration<InetAddress> inetAddr = intface.getInetAddresses(); inetAddr.hasMoreElements(); ) {
+				InetAddress nextElement = inetAddr.nextElement();
+				if (nextElement instanceof Inet4Address) {
+					return nextElement.getHostAddress();
+				}
+			}
+		}
 		return null;
 	}