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] ----