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 2d268c93cc..1df82ac74c 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 @@ -17,11 +17,11 @@ package org.springframework.integration.ip.udp; import java.io.IOException; +import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.MulticastSocket; -import java.net.NetworkInterface; import org.springframework.messaging.MessageHandler; @@ -43,6 +43,8 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler private String localAddress; + private volatile MulticastSocket multicastSocket; + /** * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port. * @param address The multicast address. @@ -136,11 +138,11 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler socket.setTimeToLive(this.timeToLive); } setSocketAttributes(socket); - if (localAddress != null) { + if (this.localAddress != null) { InetAddress whichNic = InetAddress.getByName(this.localAddress); - NetworkInterface intfce = NetworkInterface.getByInetAddress(whichNic); - socket.setNetworkInterface(intfce); + socket.setInterface(whichNic); } + this.multicastSocket = socket; } } @@ -168,4 +170,12 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler this.localAddress = localAddress; } + @Override + protected void send(DatagramPacket packet) throws Exception { + super.send(packet); + if (logger.isDebugEnabled()) { + logger.debug("Sent packet to " + this.multicastSocket.getInterface()); + } + } + } 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 29c6576550..e8d60c7c27 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 @@ -221,7 +221,7 @@ public class UnicastSendingMessageHandler extends packet = this.mapper.fromMessage(message); this.send(packet); if (logger.isDebugEnabled()) { - logger.debug("Sent packet for message " + message); + logger.debug("Sent packet for message " + message + " to " + packet.getSocketAddress()); } if (this.waitForAck) { try { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java new file mode 100644 index 0000000000..a9a49f188d --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java @@ -0,0 +1,198 @@ +/* + * 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ip.udp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.net.DatagramPacket; +import java.net.DatagramSocket; +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.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.logging.LogFactory; +import org.junit.Rule; +import org.junit.Test; + +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +/** + * @author Mark Fisher + * @author Gary Russell + * @since 2.0 + */ +public class DatagramPacketMulticastSendingHandlerTests { + + @Rule + public MulticastRule multicastRule = new MulticastRule(); + + @Test + public void verifySendMulticast() throws Exception { + MulticastSocket socket; + try { + socket = new MulticastSocket(); + } + catch (Exception e) { + return; + } + final int testPort = socket.getLocalPort(); + final String multicastAddress = this.multicastRule.getGroup(); + final String payload = "foo"; + final CountDownLatch listening = new CountDownLatch(2); + final CountDownLatch received = new CountDownLatch(2); + Runnable catcher = new Runnable() { + @Override + public void run() { + try { + byte[] buffer = new byte[8]; + DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length); + MulticastSocket socket = new MulticastSocket(testPort); + socket.setInterface(InetAddress.getByName(multicastRule.getNic())); + InetAddress group = InetAddress.getByName(multicastAddress); + socket.joinGroup(group); + listening.countDown(); + LogFactory.getLog(getClass()) + .debug(Thread.currentThread().getName() + " waiting for packet"); + socket.receive(receivedPacket); + socket.close(); + byte[] src = receivedPacket.getData(); + int length = receivedPacket.getLength(); + int offset = receivedPacket.getOffset(); + byte[] dest = new byte[length]; + System.arraycopy(src, offset, dest, 0, length); + assertEquals(payload, new String(dest)); + LogFactory.getLog(getClass()) + .debug(Thread.currentThread().getName() + " received packet"); + received.countDown(); + } + catch (Exception e) { + listening.countDown(); + e.printStackTrace(); + } + } + }; + Executor executor = Executors.newFixedThreadPool(2); + executor.execute(catcher); + executor.execute(catcher); + assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); + MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort); + handler.setLocalAddress(this.multicastRule.getNic()); + handler.handleMessage(MessageBuilder.withPayload(payload).build()); + assertTrue(received.await(10000, TimeUnit.MILLISECONDS)); + handler.stop(); + socket.close(); + } + + @Test + public void verifySendMulticastWithAcks() throws Exception { + + 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 listening = new CountDownLatch(2); + final CountDownLatch ackListening = new CountDownLatch(1); + final CountDownLatch ackSent = new CountDownLatch(2); + Runnable catcher = new Runnable() { + @Override + public void run() { + try { + byte[] buffer = new byte[1000]; + DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length); + MulticastSocket socket = new MulticastSocket(testPort); + socket.setInterface(InetAddress.getByName(multicastRule.getNic())); + socket.setSoTimeout(8000); + InetAddress group = InetAddress.getByName(multicastAddress); + socket.joinGroup(group); + listening.countDown(); + assertTrue(ackListening.await(10, TimeUnit.SECONDS)); + LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " waiting for packet"); + socket.receive(receivedPacket); + socket.close(); + byte[] src = receivedPacket.getData(); + int length = receivedPacket.getLength(); + int offset = receivedPacket.getOffset(); + byte[] dest = new byte[6]; + System.arraycopy(src, offset+length-6, dest, 0, 6); + assertEquals(payload, new String(dest)); + LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " received packet"); + DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); + mapper.setAcknowledge(true); + mapper.setLengthCheck(true); + Message message = mapper.toMessage(receivedPacket); + Object id = message.getHeaders().get(IpHeaders.ACK_ID); + byte[] ack = id.toString().getBytes(); + DatagramPacket ackPack = new DatagramPacket(ack, ack.length, + new InetSocketAddress(multicastRule.getNic(), ackPort.get())); + DatagramSocket out = new DatagramSocket(); + out.send(ackPack); + LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " sent ack to " + + ackPack.getSocketAddress()); + out.close(); + ackSent.countDown(); + socket.close(); + } + catch (Exception e) { + listening.countDown(); + e.printStackTrace(); + } + } + }; + Executor executor = Executors.newFixedThreadPool(2); + executor.execute(catcher); + executor.execute(catcher); + assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); + MulticastSendingMessageHandler handler = + new MulticastSendingMessageHandler(multicastAddress, testPort, true, true, "localhost", 0, 10000); + handler.setLocalAddress(this.multicastRule.getNic()); + handler.setMinAcksForSuccess(2); + handler.afterPropertiesSet(); + handler.start(); + waitAckListening(handler); + ackPort.set(handler.getAckPort()); + ackListening.countDown(); + handler.handleMessage(MessageBuilder.withPayload(payload).build()); + assertTrue(ackSent.await(10000, TimeUnit.MILLISECONDS)); + handler.stop(); + socket.close(); + } + + public void waitAckListening(UnicastSendingMessageHandler handler) throws InterruptedException { + int n = 0; + while (n++ < 100 && handler.getAckPort() == 0) { + Thread.sleep(100); + } + assertTrue(n < 100); + } + +} 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 2ced052504..5d76a83cc2 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 @@ -22,16 +22,12 @@ import static org.mockito.Mockito.mock; import java.net.DatagramPacket; import java.net.DatagramSocket; -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.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; @@ -46,8 +42,6 @@ import org.springframework.messaging.Message; */ public class DatagramPacketSendingHandlerTests { - private boolean noMulticast; - @Test public void verifySend() throws Exception { byte[] buffer = new byte[8]; @@ -154,147 +148,4 @@ public class DatagramPacketSendingHandlerTests { assertTrue(n < 100); } - @Test - public void verifySendMulticast() throws Exception { - 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 listening = new CountDownLatch(2); - final CountDownLatch received = new CountDownLatch(2); - Runnable catcher = new Runnable() { - @Override - public void run() { - try { - byte[] buffer = new byte[8]; - DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length); - MulticastSocket socket = new MulticastSocket(testPort); - InetAddress group = InetAddress.getByName(multicastAddress); - socket.joinGroup(group); - listening.countDown(); - LogFactory.getLog(getClass()) - .debug(Thread.currentThread().getName() + " waiting for packet"); - socket.receive(receivedPacket); - socket.close(); - byte[] src = receivedPacket.getData(); - int length = receivedPacket.getLength(); - int offset = receivedPacket.getOffset(); - byte[] dest = new byte[length]; - System.arraycopy(src, offset, dest, 0, length); - assertEquals(payload, new String(dest)); - LogFactory.getLog(getClass()) - .debug(Thread.currentThread().getName() + " received packet"); - received.countDown(); - } - catch (Exception e) { - noMulticast = true; - listening.countDown(); - e.printStackTrace(); - } - } - }; - Executor executor = Executors.newFixedThreadPool(2); - executor.execute(catcher); - executor.execute(catcher); - listening.await(10000, TimeUnit.MILLISECONDS); - if (noMulticast) { - socket.close(); - return; - } - MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort); - handler.handleMessage(MessageBuilder.withPayload(payload).build()); - assertTrue(received.await(10000, TimeUnit.MILLISECONDS)); - handler.stop(); - socket.close(); - } - - @Test - public void verifySendMulticastWithAcks() throws Exception { - - 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 listening = new CountDownLatch(2); - final CountDownLatch ackListening = new CountDownLatch(1); - final CountDownLatch ackSent = new CountDownLatch(2); - Runnable catcher = new Runnable() { - @Override - public void run() { - try { - 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); - listening.countDown(); - assertTrue(ackListening.await(10, TimeUnit.SECONDS)); - LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " waiting for packet"); - socket.receive(receivedPacket); - socket.close(); - byte[] src = receivedPacket.getData(); - int length = receivedPacket.getLength(); - int offset = receivedPacket.getOffset(); - byte[] dest = new byte[6]; - System.arraycopy(src, offset+length-6, dest, 0, 6); - assertEquals(payload, new String(dest)); - LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " received packet"); - DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); - mapper.setAcknowledge(true); - mapper.setLengthCheck(true); - Message message = mapper.toMessage(receivedPacket); - Object id = message.getHeaders().get(IpHeaders.ACK_ID); - byte[] ack = id.toString().getBytes(); - DatagramPacket ackPack = new DatagramPacket(ack, ack.length, - new InetSocketAddress("localHost", ackPort.get())); - DatagramSocket out = new DatagramSocket(); - out.send(ackPack); - out.close(); - ackSent.countDown(); - socket.close(); - } - catch (Exception e) { - noMulticast = true; - listening.countDown(); - e.printStackTrace(); - } - } - }; - Executor executor = Executors.newFixedThreadPool(2); - executor.execute(catcher); - executor.execute(catcher); - listening.await(10000, TimeUnit.MILLISECONDS); - if (this.noMulticast) { - socket.close(); - return; - } - MulticastSendingMessageHandler handler = - 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(ackSent.await(10000, TimeUnit.MILLISECONDS)); - handler.stop(); - socket.close(); - } - } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MulticastRule.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MulticastRule.java new file mode 100644 index 0000000000..e43bbf9541 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MulticastRule.java @@ -0,0 +1,108 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ip.udp; + +import java.net.InetAddress; +import java.net.MulticastSocket; + +import org.apache.commons.logging.LogFactory; +import org.junit.Assume; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; +import org.junit.runners.model.Statement; + +import org.springframework.integration.ip.util.SocketTestUtils; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + * @since 4.3 + */ +public class MulticastRule extends TestWatcher { + + public static String GROUP = "225.6.7.8"; + + private final String group; + + private final String nic; + + private boolean skip; + + public MulticastRule() { + this(GROUP); + } + + public MulticastRule(String group) { + Assert.hasText(group); + this.group = group; + System.setProperty("java.net.preferIPv4Stack", "true"); + System.setProperty("multicast.group", this.group); + try { + this.nic = checkMulticast(); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + if (this.nic != null) { + System.setProperty("multicast.local.address", this.nic); + } + } + + private String checkMulticast() throws Exception { + String nic = SocketTestUtils.chooseANic(true); + if (nic == null) { // no multicast support + this.skip = true; + return null; + } + try { + MulticastSocket socket = new MulticastSocket(); + socket.joinGroup(InetAddress.getByName(this.group)); + socket.close(); + } + catch (Exception e) { + this.skip = true; + // Ignore. Assume no Multicast - skip the test. + } + return nic; + } + + public String getGroup() { + return group; + } + + public String getNic() { + return nic; + } + + @Override + public Statement apply(Statement base, Description description) { + if (this.skip) { + LogFactory.getLog(this.getClass()).info("No Multicast support; test skipped"); + return new Statement() { + + @Override + public void evaluate() throws Throwable { + Assume.assumeTrue(false); + } + }; + } + else { + return super.apply(base, description); + } + } + +} 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 e6c43d4840..52eabe772f 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,9 +26,7 @@ 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; @@ -37,7 +35,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import org.apache.commons.logging.LogFactory; +import org.junit.Rule; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; @@ -60,6 +58,9 @@ import org.springframework.messaging.SubscribableChannel; */ public class UdpChannelAdapterTests { + @Rule + public MulticastRule multicastRule = new MulticastRule(); + @Test public void testUnicastReceiver() throws Exception { testUnicastReceiver(false, false); @@ -248,14 +249,11 @@ public class UdpChannelAdapterTests { @SuppressWarnings("unchecked") @Test public void testMulticastReceiver() throws Exception { - System.setProperty("java.net.preferIPv4Stack", "true"); QueueChannel channel = new QueueChannel(2); - MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.8", 0); + MulticastReceivingChannelAdapter adapter = + new MulticastReceivingChannelAdapter(this.multicastRule.getGroup(), 0); adapter.setOutputChannel(channel); - String nic = checkMulticast(); - if (nic == null) { - return; - } + String nic = this.multicastRule.getNic(); adapter.setLocalAddress(nic); adapter.start(); SocketTestUtils.waitListening(adapter); @@ -264,7 +262,7 @@ public class UdpChannelAdapterTests { Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); DatagramPacket packet = mapper.fromMessage(message); - packet.setSocketAddress(new InetSocketAddress("225.6.7.8", port)); + packet.setSocketAddress(new InetSocketAddress(this.multicastRule.getGroup(), port)); DatagramSocket datagramSocket = new DatagramSocket(0, Inet4Address.getByName(nic)); datagramSocket.send(packet); datagramSocket.close(); @@ -275,40 +273,21 @@ 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 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); + UnicastReceivingChannelAdapter adapter = + new MulticastReceivingChannelAdapter(this.multicastRule.getGroup(), port); adapter.setOutputChannel(channel); - String nic = checkMulticast(); - if (nic == null) { - return; - } + String nic = this.multicastRule.getNic(); adapter.setLocalAddress(nic); adapter.start(); SocketTestUtils.waitListening(adapter); - MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler("225.6.7.9", port); + MulticastSendingMessageHandler handler = + new MulticastSendingMessageHandler(this.multicastRule.getGroup(), port); handler.setLocalAddress(nic); Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); handler.handleMessage(message); 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 b0b96c7742..8c375846ac 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 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,6 +25,7 @@ import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.junit.Rule; import org.junit.Test; import org.springframework.context.ApplicationContext; @@ -50,11 +51,13 @@ import org.springframework.messaging.support.GenericMessage; * received in the other context (and written back to the console). * * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ public class UdpMulticastEndToEndTests implements Runnable { - private String testingIpText; + @Rule + public MulticastRule multicastRule = new MulticastRule(); private Message finalMessage; @@ -113,6 +116,7 @@ public class UdpMulticastEndToEndTests implements Runnable { if (!readyToReceive.await(30, TimeUnit.SECONDS)) { fail("Receiver failed to start in 30s"); } + String testingIpText; try { testingIpText = ">>>>>>> Testing IP (multicast) " + new Date(); inputChannel.send(new GenericMessage(testingIpText)); 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 26db98c6a5..be8e1e0e6f 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 @@ -1,17 +1,18 @@ + http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> + +