From b4218982dace3aae76dee64ed35b9a56a48346df Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Nov 2015 11:23:09 -0500 Subject: [PATCH] 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. --- build.gradle | 4 + .../udp/MulticastReceivingChannelAdapter.java | 17 ++- .../udp/MulticastSendingMessageHandler.java | 25 ++-- .../udp/UnicastReceivingChannelAdapter.java | 27 +++- .../ip/udp/UnicastSendingMessageHandler.java | 72 ++++++--- .../udp/DatagramPacketMessageMapperTests.java | 13 +- .../DatagramPacketSendingHandlerTests.java | 140 +++++++++++------- .../ip/udp/UdpChannelAdapterTests.java | 71 ++++++--- .../ip/udp/UdpMulticastEndToEndTests.java | 57 ++++++- .../ip/udp/UdpUnicastEndToEndTests.java | 54 +++++-- .../integration/ip/udp/testIp-in-context.xml | 10 +- .../ip/udp/testIp-in-multicast-context.xml | 8 +- .../integration/ip/udp/testIp-out-context.xml | 22 +-- .../ip/udp/testIp-out-multicast-context.xml | 24 +-- .../udp/testIp-out-within-chain-context.xml | 20 +-- .../integration/ip/util/SocketTestUtils.java | 27 ++-- src/reference/asciidoc/ip.adoc | 5 + 17 files changed, 396 insertions(+), 200 deletions(-) diff --git a/build.gradle b/build.gradle index 0333a63068..d53b48085e 100644 --- a/build.gradle +++ b/build.gradle @@ -401,6 +401,10 @@ project('spring-integration-ip') { compile project(":spring-integration-core") testCompile project(":spring-integration-stream") } + + test { + jvmArgs '-Djava.net.preferIPv4Stack=true' // Multicast on OS X + } } project('spring-integration-jdbc') { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java index e93dd3b798..9b659467ed 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 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. @@ -26,17 +26,17 @@ import org.springframework.messaging.MessagingException; /** * Channel adapter that joins a multicast group and receives incoming packets and * sends them to an output channel. - * + * * @author Gary Russell * @since 2.0 */ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAdapter { - private String group; + private final String group; /** - * Constructs a MulticastReceivingChannelAdapter that listens for packets on the + * Constructs a MulticastReceivingChannelAdapter that listens for packets on the * specified multichannel address (group) and port. * @param group The multichannel address. * @param port The port. @@ -47,7 +47,7 @@ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAda } /** - * Constructs a MulticastReceivingChannelAdapter that listens for packets on the + * Constructs a MulticastReceivingChannelAdapter that listens for packets on the * specified multichannel address (group) and port. Enables setting the lengthCheck * option, which expects a length to precede the incoming packets. * @param group The multichannel address. @@ -63,15 +63,16 @@ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAda protected synchronized DatagramSocket getSocket() { if (this.getTheSocket() == null) { try { - MulticastSocket socket = new MulticastSocket(this.getPort()); + int port = getPort(); + MulticastSocket socket = port == 0 ? new MulticastSocket() : new MulticastSocket(port); String localAddress = this.getLocalAddress(); if (localAddress != null) { InetAddress whichNic = InetAddress.getByName(localAddress); socket.setInterface(whichNic); } - this.setSocketAttributes(socket); + setSocketAttributes(socket); socket.joinGroup(InetAddress.getByName(this.group)); - this.setSocket(socket); + setSocket(socket); } catch (IOException e) { throw new MessagingException("failed to create DatagramSocket", e); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index f02477d2f0..2d268c93cc 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -111,20 +111,26 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler if (this.getTheSocket() == null) { MulticastSocket socket; if (this.isAcknowledge()) { - if (logger.isDebugEnabled()) { - logger.debug("Listening for acks on port: " + this.getAckPort()); - } + int ackPort = this.getAckPort(); if (localAddress == null) { - socket = new MulticastSocket(this.getAckPort()); - } else { - InetAddress whichNic = InetAddress.getByName(this.localAddress); - socket = new MulticastSocket(new InetSocketAddress(whichNic, this.getAckPort())); + socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort); } - if (this.getSoReceiveBufferSize() > 0) { + else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort)); + } + if (getSoReceiveBufferSize() > 0) { socket.setReceiveBufferSize(this.getSoReceiveBufferSize()); } - } else { + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + socket.getLocalPort()); + } + setSocket(socket); + updateAckAddress(); + } + else { socket = new MulticastSocket(); + setSocket(socket); } if (this.timeToLive >= 0) { socket.setTimeToLive(this.timeToLive); @@ -135,7 +141,6 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler NetworkInterface intfce = NetworkInterface.getByInetAddress(whichNic); socket.setNetworkInterface(intfce); } - this.setSocket(socket); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index aa50f5e9ef..3fb1eff4b7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -79,6 +79,16 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece return true; } + @Override + public int getPort() { + if (this.socket == null) { + return super.getPort(); + } + else { + return this.socket.getLocalPort(); + } + } + @Override protected void onInit() { super.onInit(); @@ -87,11 +97,13 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece @Override public void run() { + getSocket(); + if (logger.isDebugEnabled()) { logger.debug("UDP Receiver running on port:" + this.getPort()); } - this.setListening(true); + setListening(true); // Do as little as possible here so we can loop around and catch the next packet. // Just schedule the packet for processing. @@ -122,7 +134,8 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece String ackAddress = ((String) headers.get(IpHeaders.ACK_ADDRESS)).trim(); Matcher mat = addressPattern.matcher(ackAddress); if (!mat.matches()) { - throw new MessagingException(message, "Ack requested but could not decode acknowledgment address:" + ackAddress); + throw new MessagingException(message, + "Ack requested but could not decode acknowledgment address: " + ackAddress); } String host = mat.group(1); int port = Integer.parseInt(mat.group(2)); @@ -140,7 +153,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece out.close(); } catch (IOException e) { - throw new MessagingException(message, "Failed to send acknowledgment", e); + throw new MessagingException(message, "Failed to send acknowledgment to: " + ackAddress, e); } } @@ -210,11 +223,13 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece try { DatagramSocket socket = null; String localAddress = this.getLocalAddress(); + int port = super.getPort(); if (localAddress == null) { - socket = new DatagramSocket(this.getPort()); - } else { + socket = port == 0 ? new DatagramSocket() : new DatagramSocket(port); + } + else { InetAddress whichNic = InetAddress.getByName(localAddress); - socket = new DatagramSocket(this.getPort(), whichNic); + socket = new DatagramSocket(new InetSocketAddress(whichNic, port)); } setSocketAttributes(socket); this.socket = socket; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index 176b624763..10aba4a1b7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -20,6 +20,7 @@ import java.io.IOException; import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.SocketException; import java.util.Collections; import java.util.HashMap; @@ -66,6 +67,8 @@ public class UnicastSendingMessageHandler extends private volatile boolean acknowledge = false; + private volatile String ackHost; + private volatile int ackPort; private volatile int ackTimeout = 5000; @@ -158,6 +161,7 @@ public class UnicastSendingMessageHandler extends this.waitForAck = acknowledge; this.mapper.setAcknowledge(acknowledge); this.mapper.setAckAddress(ackHost + ":" + ackPort); + this.ackHost = ackHost; this.ackPort = ackPort; if (ackTimeout > 0) { this.ackTimeout = ackTimeout; @@ -185,6 +189,7 @@ public class UnicastSendingMessageHandler extends }); this.taskExecutor = executor; } + startAckThread(); } } @@ -203,19 +208,7 @@ public class UnicastSendingMessageHandler extends MessageDeliveryException { if (this.acknowledge) { Assert.state(this.isRunning(), "When 'acknowlege' is enabled, adapter must be running"); - if (!this.ackThreadRunning) { - synchronized(this) { - if (!this.ackThreadRunning) { - ackLatch = new CountDownLatch(1); - this.taskExecutor.execute(this); - try { - ackLatch.await(10000, TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } - } - } + startAckThread(); } CountDownLatch countdownLatch = null; String messageId = message.getHeaders().getId().toString(); @@ -227,13 +220,16 @@ public class UnicastSendingMessageHandler extends } packet = this.mapper.fromMessage(message); this.send(packet); - logger.debug("Sent packet for message " + message); + if (logger.isDebugEnabled()) { + logger.debug("Sent packet for message " + message); + } if (this.waitForAck) { try { if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) { throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis"); } - } catch (InterruptedException e) { + } + catch (InterruptedException e) { Thread.currentThread().interrupt(); } } @@ -256,6 +252,23 @@ public class UnicastSendingMessageHandler extends } } + public void startAckThread() { + if (!this.ackThreadRunning) { + synchronized(this) { + if (!this.ackThreadRunning) { + ackLatch = new CountDownLatch(1); + this.taskExecutor.execute(this); + try { + ackLatch.await(10000, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + } + } + protected void send(DatagramPacket packet) throws Exception { DatagramSocket socket = this.getSocket(); packet.setSocketAddress(this.getDestinationAddress()); @@ -273,19 +286,22 @@ public class UnicastSendingMessageHandler extends protected synchronized DatagramSocket getSocket() throws IOException { if (this.socket == null) { if (acknowledge) { - if (logger.isDebugEnabled()) { - logger.debug("Listening for acks on port: " + ackPort); - } if (localAddress == null) { - this.socket = new DatagramSocket(this.ackPort); - } else { + this.socket = this.ackPort == 0 ? new DatagramSocket() : new DatagramSocket(this.ackPort); + } + else { InetAddress whichNic = InetAddress.getByName(this.localAddress); - this.socket = new DatagramSocket(this.ackPort, whichNic); + this.socket = new DatagramSocket(new InetSocketAddress(whichNic, this.ackPort)); } if (this.soReceiveBufferSize > 0) { socket.setReceiveBufferSize(this.soReceiveBufferSize); } - } else { + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + getAckPort()); + } + updateAckAddress(); + } + else { this.socket = new DatagramSocket(); } setSocketAttributes(this.socket); @@ -293,6 +309,10 @@ public class UnicastSendingMessageHandler extends return this.socket; } + protected void updateAckAddress() { + this.mapper.setAckAddress(this.ackHost + ":" + getAckPort()); + } + /** * @see java.net.Socket#setReceiveBufferSize(int) * @see DatagramSocket#setReceiveBufferSize(int) @@ -336,7 +356,13 @@ public class UnicastSendingMessageHandler extends * @return the ackPort */ public int getAckPort() { - return ackPort; + DatagramSocket socket = this.socket; + if (this.ackPort == 0 && socket != null) { + return socket.getLocalPort(); + } + else { + return this.ackPort; + } } /** diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java index a9e683140b..5ead730e69 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java +++ b/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 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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java index cad89a4f7d..2ced052504 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java +++ b/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 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 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(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index aa647e694c..e6c43d4840 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/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 exceptionHolder = new AtomicReference(); - 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 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 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 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 message = MessageBuilder.withPayload("ABCD".getBytes()).build(); DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); 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 b3af7223dc..b0b96c7742 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 @@ -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 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(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java index 229e6a9b32..2ad7786b2f 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java +++ b/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(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml index 51c84bf259..669cdd03c6 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml +++ b/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"> - + - - + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-multicast-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-multicast-context.xml index 1a8beae95d..26db98c6a5 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-multicast-context.xml +++ b/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"> - - + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml index 14591693fe..e6598fe338 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml +++ b/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"> + + @@ -26,13 +26,13 @@ ref="testIp" method="testIp"/> - diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml index 96e938417a..35cf268ada 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml +++ b/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"> + + @@ -26,18 +26,18 @@ ref="testIp" method="testIp"/> - - + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-within-chain-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-within-chain-context.xml index b2e696b4e0..88672c2df1 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-within-chain-context.xml +++ b/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"> + + @@ -28,11 +28,11 @@ diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java index 4b7d029d06..90aa3a5dd0 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java +++ b/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 interfaces = NetworkInterface.getNetworkInterfaces(); -// while (interfaces.hasMoreElements()) { -// NetworkInterface intface = interfaces.nextElement(); -// if (intface.isLoopback() || (multicast && !intface.supportsMulticast())) -// continue; -// Enumeration inet = intface.getInetAddresses(); -// if (!inet.hasMoreElements()) -// continue; -// String address = inet.nextElement().getHostAddress(); -// return address; -// } + Enumeration interfaces = NetworkInterface.getNetworkInterfaces(); + while (interfaces.hasMoreElements()) { + NetworkInterface intface = interfaces.nextElement(); + if (intface.isLoopback() || (multicast && !intface.supportsMulticast())) { + continue; + } + for (Enumeration inetAddr = intface.getInetAddresses(); inetAddr.hasMoreElements(); ) { + InetAddress nextElement = inetAddr.nextElement(); + if (nextElement instanceof Inet4Address) { + return nextElement.getHostAddress(); + } + } + } return null; } diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 733379c2b9..47720afa79 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -53,6 +53,9 @@ This enables the receiving side to verify the length of the packet received. If a receiving system uses a buffer that is too short the contain the packet, the packet can be truncated. The length header provides a mechanism to detect this. +Starting with _version 4.3_, the `port` can be set to `0`, in which case the Operating System chooses the port; the +chosen port can be discovered by invoking `getPort()` after the adapter is started and `isListening()` returns `true`. + [source,xml] ----