From 9fef40f179388bf7637f5397e429d901fef804da Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 30 May 2010 18:54:50 +0000 Subject: [PATCH] INT-1149 Support the specification of a network interface via local-address attribute on inbound endpoints and outbound UDP adapter. Docs to follow. --- ...ternetProtocolReceivingChannelAdapter.java | 10 + ...InternetProtocolSendingMessageHandler.java | 187 ++++++------ .../integration/ip/CommonSocketOptions.java | 33 +- .../ip/config/IpAdapterParserUtils.java | 3 + .../ip/tcp/SimpleTcpNetInboundGateway.java | 7 + .../ip/tcp/TcpNetReceivingChannelAdapter.java | 26 +- .../ip/tcp/TcpNetSendingMessageHandler.java | 4 + .../ip/tcp/TcpNioReceivingChannelAdapter.java | 27 +- .../ip/tcp/TcpNioSendingMessageHandler.java | 3 + .../udp/MulticastReceivingChannelAdapter.java | 5 +- .../udp/MulticastSendingMessageHandler.java | 282 ++++++++++-------- .../udp/UnicastReceivingChannelAdapter.java | 13 +- .../ip/udp/UnicastSendingMessageHandler.java | 81 +++-- .../ip/config/spring-integration-ip-2.0.xsd | 12 + .../ip/config/ParserUnitTests-context.xml | 7 +- .../ip/config/ParserUnitTests.java | 6 + .../integration/ip/tcp/MultiClientTests.java | 13 +- .../tcp/TcpReceivingChannelAdapterTests.java | 27 +- .../DatagramPacketSendingHandlerTests.java | 57 ++-- .../integration/ip/udp/MultiClientTests.java | 17 +- .../ip/udp/UdpChannelAdapterTests.java | 133 +++++++++ .../ip/udp/UdpMulticastEndToEndTests.java | 5 + .../integration/ip/util/SocketUtils.java | 43 +++ 23 files changed, 667 insertions(+), 334 deletions(-) create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 6ce2f9dc26..1cd1a4c0a5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -44,6 +44,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter protected volatile boolean listening; + protected volatile String localAddress; + public AbstractInternetProtocolReceivingChannelAdapter(int port) { this.port = port; @@ -101,4 +103,12 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter return listening; } + public String getLocalAddress() { + return localAddress; + } + + public void setLocalAddress(String localAddress) { + this.localAddress = localAddress; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index 36f76c4404..92ce8ee4e9 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -1,94 +1,93 @@ -/* - * Copyright 2002-2010 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; - -import java.net.DatagramSocket; -import java.net.InetSocketAddress; -import java.net.Socket; -import java.net.SocketAddress; -import java.util.concurrent.ExecutorService; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.integration.message.MessageHandler; -import org.springframework.util.Assert; - -/** - * Base class for all TCP/UDP MessageHandlers. - * - * @author Gary Russell - * @since 2.0 - */ -public abstract class AbstractInternetProtocolSendingMessageHandler implements MessageHandler, CommonSocketOptions { - - protected final Log logger = LogFactory.getLog(getClass()); - - protected final SocketAddress destinationAddress; - - protected final String host; - - protected final int port; - - protected volatile int soSendBufferSize = -1; - - protected volatile int soTimeout = -1; - - protected volatile ExecutorService executorService; - - - public AbstractInternetProtocolSendingMessageHandler(String host, int port) { - Assert.notNull(host, "host must not be null"); - this.destinationAddress = new InetSocketAddress(host, port); - this.host = host; - this.port = port; - } - - - /** - * @see Socket#setSoTimeout(int) - * @see DatagramSocket#setSoTimeout(int) - * @param timeout - */ - public void setSoTimeout(int timeout) { - this.soTimeout = timeout; - } - - /** - * @see Socket#setReceiveBufferSize(int) - * @see DatagramSocket#setReceiveBufferSize(int) - * @param size - */ - public void setSoReceiveBufferSize(int size) { - } - - /** - * @see Socket#setSendBufferSize(int) - * @see DatagramSocket#setSendBufferSize(int) - * @param size - */ - public void setSoSendBufferSize(int size) { - this.soSendBufferSize = size; - } - - /** - * @return the port - */ - public int getPort() { - return port; - } - -} +/* + * Copyright 2002-2010 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; + +import java.net.DatagramSocket; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketAddress; +import java.util.concurrent.ExecutorService; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.message.MessageHandler; +import org.springframework.util.Assert; + +/** + * Base class for all TCP/UDP MessageHandlers. + * + * @author Gary Russell + * @since 2.0 + */ +public abstract class AbstractInternetProtocolSendingMessageHandler implements MessageHandler, CommonSocketOptions { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final SocketAddress destinationAddress; + + protected final String host; + + protected final int port; + + protected volatile int soSendBufferSize = -1; + + protected volatile int soTimeout = -1; + + protected volatile ExecutorService executorService; + + public AbstractInternetProtocolSendingMessageHandler(String host, int port) { + Assert.notNull(host, "host must not be null"); + this.destinationAddress = new InetSocketAddress(host, port); + this.host = host; + this.port = port; + } + + + /** + * @see Socket#setSoTimeout(int) + * @see DatagramSocket#setSoTimeout(int) + * @param timeout + */ + public void setSoTimeout(int timeout) { + this.soTimeout = timeout; + } + + /** + * @see Socket#setReceiveBufferSize(int) + * @see DatagramSocket#setReceiveBufferSize(int) + * @param size + */ + public void setSoReceiveBufferSize(int size) { + } + + /** + * @see Socket#setSendBufferSize(int) + * @see DatagramSocket#setSendBufferSize(int) + * @param size + */ + public void setSoSendBufferSize(int size) { + this.soSendBufferSize = size; + } + + /** + * @return the port + */ + public int getPort() { + return port; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java index 28f7f230c6..4de8044e9c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java @@ -22,10 +22,37 @@ package org.springframework.integration.ip; */ public interface CommonSocketOptions { - void setSoTimeout(int soTimeout); + /** + * @see Socket#setSoTimeout(int) + * @see DatagramSocket#setSoTimeout(int) + * @param timeout + */ + public void setSoTimeout(int soTimeout); - void setSoReceiveBufferSize(int soReceiveBufferSize); + /** + * @see Socket#setReceiveBufferSize(int) + * @see DatagramSocket#setReceiveBufferSize(int) + * @param size + */ + public void setSoReceiveBufferSize(int soReceiveBufferSize); - void setSoSendBufferSize(int soSendBufferSize); + /** + * @see Socket#setSendBufferSize(int) + * @see DatagramSocket#setSendBufferSize(int) + * @param size + */ + public void setSoSendBufferSize(int soSendBufferSize); + + /** + * On a multi-homed system, specifies the ip address of the network interface used to communicate. + * For inbound adapters and gateways, specifies the interface used to listed for incoming connections. + * If omitted, the endpoint will listen on all available adapters. For the UDP multicast outbound adapter + * specifies the interface to which multicast packets will be sent. For UDP unicast and multicast + * adapters, specifies which interface to which the acknowledgment socket will be bound. Does not + * apply to TCP outbound adapters and gateways. + * + * @param localAddress + */ + public void setLocalAddress(String localAddress); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index c24b013b2d..937ecb5e45 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -88,6 +88,8 @@ public abstract class IpAdapterParserUtils { static final String SO_TRAFFIC_CLASS = "so-traffic-class"; static final String CLOSE = "close"; + + static final String LOCAL_ADDRESS = "local-address"; /** @@ -265,6 +267,7 @@ public abstract class IpAdapterParserUtils { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_TIMEOUT); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_RECEIVE_BUFFER_SIZE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_SEND_BUFFER_SIZE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, LOCAL_ADDRESS); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java index 7def8795aa..ef59f2242e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java @@ -66,6 +66,8 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { protected boolean close; + protected String localAddress; + @Override protected void doStart() { super.doStart(); @@ -91,6 +93,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { this.delegate.setTaskScheduler(getTaskScheduler()); this.delegate.setCustomSocketReaderClassName(customSocketReaderClassName); this.delegate.setClose(close); + this.delegate.setLocalAddress(localAddress); super.onInit(); } @@ -206,6 +209,10 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { return delegate.isListening(); } + public void setLocalAddress(String localAddress) { + this.localAddress = localAddress; + } + private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter { /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java index c33cc6ae76..397354449b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.net.InetAddress; import java.net.ServerSocket; import java.net.Socket; @@ -37,6 +38,7 @@ public class TcpNetReceivingChannelAdapter extends AbstractTcpReceivingChannelAdapter { protected ServerSocket serverSocket; + protected Class customSocketReaderClass; /** * Constructs a TcpNetReceivingChannelAdapter that listens on the provided port. @@ -55,11 +57,17 @@ public class TcpNetReceivingChannelAdapter extends */ @Override protected void server() { - while (active) { + while (this.active) { try { - serverSocket = ServerSocketFactory.getDefault() - .createServerSocket(port, Math.abs(poolSize)); - listening = true; + if (this.localAddress == null) { + this.serverSocket = ServerSocketFactory.getDefault() + .createServerSocket(this.port, Math.abs(this.poolSize)); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + this.serverSocket = ServerSocketFactory.getDefault() + .createServerSocket(port, Math.abs(poolSize), whichNic); + } + this.listening = true; while (true) { final Socket socket = serverSocket.accept(); setSocketOptions(socket); @@ -69,14 +77,14 @@ public class TcpNetReceivingChannelAdapter extends }}); } } catch (IOException e) { - if (serverSocket != null) { + if (this.serverSocket != null) { try { - serverSocket.close(); + this.serverSocket.close(); } catch (IOException e1) {} } - listening = false; - serverSocket = null; - if (active) { + this.listening = false; + this.serverSocket = null; + if (this.active) { logger.error("Error on ServerSocket", e); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java index 3d67f0b1b5..f3fa55a82c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java @@ -95,4 +95,8 @@ public class TcpNetSendingMessageHandler extends this.writer.doClose(); this.writer = null; } + + public void setLocalAddress(String localAddress) { + logger.warn("localAddress not used on tcp outbound endpoints"); + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index f6d59e2344..81152291a4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.Socket; import java.net.SocketException; @@ -65,22 +66,28 @@ public class TcpNioReceivingChannelAdapter extends @Override protected void server() { try { - serverChannel = ServerSocketChannel.open(); - listening = true; - serverChannel.configureBlocking(false); - serverChannel.socket().bind(new InetSocketAddress(port), - Math.abs(poolSize)); + this.serverChannel = ServerSocketChannel.open(); + this.listening = true; + this.serverChannel.configureBlocking(false); + if (this.localAddress == null) { + this.serverChannel.socket().bind(new InetSocketAddress(this.port), + Math.abs(this.poolSize)); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + this.serverChannel.socket().bind(new InetSocketAddress(whichNic, this.port), + Math.abs(this.poolSize)); + } final Selector selector = Selector.open(); - serverChannel.register(selector, SelectionKey.OP_ACCEPT); - doSelect(serverChannel, selector); + this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); + doSelect(this.serverChannel, selector); } catch (IOException e) { try { serverChannel.close(); } catch (IOException e1) { } - listening = false; - serverChannel = null; - if (active) { + this.listening = false; + this.serverChannel = null; + if (this.active) { logger.error("Error on ServerSocketChannel", e); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java index 1f174e16d1..cf45486b25 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java @@ -102,4 +102,7 @@ public class TcpNioSendingMessageHandler extends this.buffsPerConnection = buffsPerConnection; } + public void setLocalAddress(String localAddress) { + logger.warn("localAddress not used on tcp outbound endpoints"); + } } 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 029b106927..1746ec0ab9 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 @@ -59,12 +59,15 @@ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAda this.group = group; } - @Override protected synchronized DatagramSocket getSocket() { if (this.socket == null) { try { MulticastSocket socket = new MulticastSocket(this.port); + if (localAddress != null) { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + socket.setInterface(whichNic); + } socket.setSoTimeout(this.soTimeout); if (this.soReceiveBufferSize > 0) { socket.setReceiveBufferSize(this.soReceiveBufferSize); 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 8532356731..90eb168a97 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 @@ -1,126 +1,156 @@ -/* - * Copyright 2002-2010 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.io.IOException; -import java.net.DatagramSocket; -import java.net.MulticastSocket; - -import org.springframework.integration.message.MessageHandler; - -/** - * A {@link MessageHandler} implementation that maps a Message into - * a UDP datagram packet and sends that to the specified multicast address - * (224.0.0.0 to 239.255.255.255) and port. - * - * The only difference between this and its super class is the - * ability to specify how many acknowledgments are required to - * determine success. - * - * @author Gary Russell - * @since 2.0 - */ -public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler { - - protected int timeToLive = -1; - - - /** - * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port. - * @param address The multicast address. - * @param port The port. - */ - public MulticastSendingMessageHandler(String address, int port) { - super(address, port); - } - - /** - * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port - * and enables setting the lengthCheck option (if set, a length is prepended to the packet and checked - * at the destination). - * @param address The multicast address. - * @param port The port. - * @param lengthCheck Enable the lengthCheck option. - */ - public MulticastSendingMessageHandler(String address, int port, boolean lengthCheck) { - super(address, port, lengthCheck); - } - - - /** - * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port - * and enables setting the acknowledge option, where the destination sends a receipt acknowledgment. - * @param address The multicast address. - * @param port The port. - * @param acknowledge Whether or not acknowledgments are required. - * @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true. - * @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true. - * @param ackTimeout How long to wait (milliseconds) for an acknowledgment. - */ - public MulticastSendingMessageHandler(String address, int port, - boolean acknowledge, String ackHost, int ackPort, int ackTimeout) { - super(address, port, acknowledge, ackHost, ackPort, ackTimeout); - } - - /** - * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port - * and enables setting the acknowledge option, where the destination sends a receipt acknowledgment. - * @param address The multicast address. - * @param port The port. - * @param lengthCheck Enable the lengthCheck option. - * @param acknowledge Whether or not acknowledgments are required. - * @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true. - * @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true. - * @param ackTimeout How long to wait (milliseconds) for an acknowledgment. - */ - public MulticastSendingMessageHandler(String address, int port, - boolean lengthCheck, boolean acknowledge, String ackHost, - int ackPort, int ackTimeout) { - super(address, port, lengthCheck, acknowledge, ackHost, ackPort, ackTimeout); - } - - - /** - * If acknowledge = true; how many acks needed for success. - * @param minAcksForSuccess - */ - public void setMinAcksForSuccess(int minAcksForSuccess) { - this.ackCounter = minAcksForSuccess; - } - - /** - * Set the underlying {@link MulticastSocket} time to live property. - * @param timeToLive {@link MulticastSocket#setTimeToLive(int)} - */ - public void setTimeToLive(int timeToLive) { - this.timeToLive = timeToLive; - } - - protected synchronized DatagramSocket getSocket() throws IOException { - if (this.socket == null) { - MulticastSocket socket = new MulticastSocket(); - if (this.timeToLive >= 0) { - socket.setTimeToLive(this.timeToLive); - } - socket.setLoopbackMode(true); // disable loopback to the local port - setSocketAttributes(socket); - this.socket = socket; - } - return this.socket; - } - -} +/* + * Copyright 2002-2010 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.io.IOException; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.MulticastSocket; +import java.net.NetworkInterface; +import java.net.SocketAddress; + +import org.springframework.integration.message.MessageHandler; + +/** + * A {@link MessageHandler} implementation that maps a Message into + * a UDP datagram packet and sends that to the specified multicast address + * (224.0.0.0 to 239.255.255.255) and port. + * + * The only difference between this and its super class is the + * ability to specify how many acknowledgments are required to + * determine success. + * + * @author Gary Russell + * @since 2.0 + */ +public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler { + + protected int timeToLive = -1; + + protected String localAddress; + + /** + * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port. + * @param address The multicast address. + * @param port The port. + */ + public MulticastSendingMessageHandler(String address, int port) { + super(address, port); + } + + /** + * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port + * and enables setting the lengthCheck option (if set, a length is prepended to the packet and checked + * at the destination). + * @param address The multicast address. + * @param port The port. + * @param lengthCheck Enable the lengthCheck option. + */ + public MulticastSendingMessageHandler(String address, int port, boolean lengthCheck) { + super(address, port, lengthCheck); + } + + + /** + * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port + * and enables setting the acknowledge option, where the destination sends a receipt acknowledgment. + * @param address The multicast address. + * @param port The port. + * @param acknowledge Whether or not acknowledgments are required. + * @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true. + * @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true. + * @param ackTimeout How long to wait (milliseconds) for an acknowledgment. + */ + public MulticastSendingMessageHandler(String address, int port, + boolean acknowledge, String ackHost, int ackPort, int ackTimeout) { + super(address, port, acknowledge, ackHost, ackPort, ackTimeout); + } + + /** + * Constructs a MulticastSendingMessageHandler to send data to the multicast address/port + * and enables setting the acknowledge option, where the destination sends a receipt acknowledgment. + * @param address The multicast address. + * @param port The port. + * @param lengthCheck Enable the lengthCheck option. + * @param acknowledge Whether or not acknowledgments are required. + * @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true. + * @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true. + * @param ackTimeout How long to wait (milliseconds) for an acknowledgment. + */ + public MulticastSendingMessageHandler(String address, int port, + boolean lengthCheck, boolean acknowledge, String ackHost, + int ackPort, int ackTimeout) { + super(address, port, lengthCheck, acknowledge, ackHost, ackPort, ackTimeout); + } + + @Override + protected synchronized DatagramSocket getSocket() throws IOException { + if (this.socket == null) { + MulticastSocket socket; + if (acknowledge) { + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + ackPort); + } + if (localAddress == null) { + socket = new MulticastSocket(this.ackPort); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + socket = new MulticastSocket(new InetSocketAddress(whichNic, this.ackPort)); + } + if (this.soReceiveBufferSize > 0) { + socket.setReceiveBufferSize(this.soReceiveBufferSize); + } + } else { + socket = new MulticastSocket(); + } + if (this.timeToLive >= 0) { + socket.setTimeToLive(this.timeToLive); + } + setSocketAttributes(socket); + if (localAddress != null) { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + NetworkInterface intfce = NetworkInterface.getByInetAddress(whichNic); + socket.setNetworkInterface(intfce); + } + this.socket = socket; + } + return this.socket; + } + + + /** + * If acknowledge = true; how many acks needed for success. + * @param minAcksForSuccess + */ + public void setMinAcksForSuccess(int minAcksForSuccess) { + this.ackCounter = minAcksForSuccess; + } + + /** + * Set the underlying {@link MulticastSocket} time to live property. + * @param timeToLive {@link MulticastSocket#setTimeToLive(int)} + */ + public void setTimeToLive(int timeToLive) { + this.timeToLive = timeToLive; + } + + public void setLocalAddress(String localAddress) { + this.localAddress = localAddress; + } + +} 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 6a4e3d583b..7b2a4fecbb 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 @@ -19,6 +19,7 @@ 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.SocketException; import java.net.SocketTimeoutException; @@ -161,6 +162,8 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece Message message = null; try { message = mapper.toMessage(packet); + if (logger.isDebugEnabled()) + logger.debug("Received:" + message); } catch (Exception e) { logger.error("Failed to map packet to message ", e); @@ -186,13 +189,19 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece protected synchronized DatagramSocket getSocket() { if (this.socket == null) { try { - this.socket = new DatagramSocket(this.port); + if (localAddress == null) { + this.socket = new DatagramSocket(this.port); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + this.socket = new DatagramSocket(this.port, whichNic); + } + this.socket.setSoTimeout(this.soTimeout); if (this.soReceiveBufferSize > 0) { this.socket.setReceiveBufferSize(this.soReceiveBufferSize); } } - catch (SocketException e) { + 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/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index a4c970d931..c4d690c0b0 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 @@ -17,9 +17,9 @@ package org.springframework.integration.ip.udp; import java.io.IOException; -import java.net.BindException; import java.net.DatagramPacket; import java.net.DatagramSocket; +import java.net.InetAddress; import java.net.Socket; import java.net.SocketException; import java.util.Collections; @@ -63,6 +63,8 @@ public class UnicastSendingMessageHandler extends * If true adds headers to instruct receiving adapter to return an ack. */ protected volatile boolean waitForAck = false; + + protected volatile boolean acknowledge = false; protected volatile int ackPort; @@ -73,13 +75,16 @@ public class UnicastSendingMessageHandler extends protected volatile Map ackControl = Collections .synchronizedMap(new HashMap()); - protected volatile DatagramSocket ackSocket; - protected volatile Exception fatalException; protected int soReceiveBufferSize = -1; + protected String localAddress; + + private CountDownLatch ackLatch; + private boolean ackThreadRunning; + /** * Basic constructor; no reliability; no acknowledgment. * @param host Destination host. @@ -157,6 +162,7 @@ public class UnicastSendingMessageHandler extends } if (acknowledge) { Assert.hasLength(ackHost); + this.acknowledge = true; this.executorService = Executors .newSingleThreadExecutor(new ThreadFactory() { private AtomicInteger n = new AtomicInteger(); @@ -167,13 +173,25 @@ public class UnicastSendingMessageHandler extends return thread; } }); - this.executorService.execute(this); } } public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { + if (this.acknowledge) { + if (!this.ackThreadRunning) { + synchronized(this) { + if (!this.ackThreadRunning) { + ackLatch = new CountDownLatch(1); + this.executorService.execute(this); + try { + ackLatch.await(10000, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { } + } + } + } + } CountDownLatch countdownLatch = null; String messageId = message.getHeaders().getId().toString(); try { @@ -187,7 +205,7 @@ public class UnicastSendingMessageHandler extends } packet = this.mapper.fromMessage(message); this.send(packet); - logger.debug("Sent packet for message id " + message.getHeaders().getId()); + logger.debug("Sent packet for message " + message); if (this.waitForAck) { if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) { throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis"); @@ -219,7 +237,22 @@ public class UnicastSendingMessageHandler extends protected synchronized DatagramSocket getSocket() throws IOException { if (this.socket == null) { - this.socket = new DatagramSocket(); + if (acknowledge) { + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + ackPort); + } + if (localAddress == null) { + this.socket = new DatagramSocket(this.ackPort); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + this.socket = new DatagramSocket(this.ackPort, whichNic); + } + if (this.soReceiveBufferSize > 0) { + socket.setReceiveBufferSize(this.soReceiveBufferSize); + } + } else { + this.socket = new DatagramSocket(); + } setSocketAttributes(this.socket); } return this.socket; @@ -238,18 +271,12 @@ public class UnicastSendingMessageHandler extends * Process acknowledgments, if requested. */ public void run() { - Exception fatalException = null; try { - if (logger.isDebugEnabled()) { - logger.debug("Listening for acks on port: " + ackPort); - } - this.ackSocket = new DatagramSocket(this.ackPort); - if (this.soReceiveBufferSize > 0) { - ackSocket.setReceiveBufferSize(this.soReceiveBufferSize); - } + this.ackThreadRunning = true; + ackLatch.countDown(); DatagramPacket ackPack = new DatagramPacket(new byte[100], 100); while(true) { - this.ackSocket.receive(ackPack); + this.getSocket().receive(ackPack); String id = new String(ackPack.getData(), ackPack.getOffset(), ackPack.getLength()); if (logger.isDebugEnabled()) { logger.debug("Received ack for " + id + " from " + ackPack.getAddress().getHostAddress()); @@ -261,20 +288,12 @@ public class UnicastSendingMessageHandler extends } } catch (IOException e) { - logger.error("Error on UDP Acknowledge thread" + e.getMessage()); - fatalException = e; + if (this.socket != null && !this.socket.isClosed()) { + logger.error("Error on UDP Acknowledge thread:" + e.getMessage()); + } } finally { - if (this.ackSocket != null) { - this.ackSocket.close(); - } - if (fatalException instanceof BindException) { - logger.fatal("Failed to bind to acknowledge port: " + ackPort); - this.fatalException = fatalException; - } - else { - this.executorService.execute(this); - } + this.ackThreadRunning = false; } } @@ -291,10 +310,9 @@ public class UnicastSendingMessageHandler extends } public void shutDown() { - DatagramSocket socket = this.ackSocket; - this.ackSocket = null; if (socket != null) { socket.close(); + socket = null; } } @@ -306,4 +324,9 @@ public class UnicastSendingMessageHandler extends this.soReceiveBufferSize = size; } + public void setLocalAddress(String localAddress) { + this.localAddress = localAddress; + } + + } diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index de89190af5..a498545fc1 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -245,6 +245,18 @@ receive the next message. + + + +On a multi-homed system, specifies the ip address of the network interface used to communicate. +For inbound adapters and gateways, specifies the interface used to listed for incoming connections. +If omitted, the endpoint will listen on all available adapters. For the UDP multicast outbound adapter +specifies the interface to which multicast packets will be sent. For UDP unicast and multicast +adapters, specifies which interface to which the acknowledgment socket will be bound. Does not +apply to TCP outbound adapters and gateways. + + + \ No newline at end of file diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 5385d938ae..1be330434c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -26,6 +26,7 @@ so-receive-buffer-size="30" so-send-buffer-size="31" so-timeout="32" + local-address="127.0.0.1" /> message = channel.receive(2000); assertNotNull(message); @@ -75,7 +76,7 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - waitListening(adapter); + SocketUtils.waitListening(adapter); SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice Message message = channel.receive(4000); assertNotNull(message); @@ -101,8 +102,9 @@ public class TcpReceivingChannelAdapterTests { ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); + SocketUtils.setLocalNicIfPossible(adapter); adapter.start(); - waitListening(adapter); + SocketUtils.waitListening(adapter); SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice Message message = channel.receive(2000); assertNotNull(message); @@ -130,7 +132,7 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - waitListening(adapter); + SocketUtils.waitListening(adapter); SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice Message message = channel.receive(2000); assertNotNull(message); @@ -160,7 +162,7 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - waitListening(adapter); + SocketUtils.waitListening(adapter); CountDownLatch latch = new CountDownLatch(1); SocketUtils.testSendCrLfSingle(port, latch); Message message = channel.receive(5000); @@ -195,7 +197,7 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - waitListening(adapter); + SocketUtils.waitListening(adapter); CountDownLatch latch = new CountDownLatch(1); SocketUtils.testSendCrLfSingle(port, latch); Message message = channel.receive(2000); @@ -210,18 +212,5 @@ public class TcpReceivingChannelAdapterTests { new String((byte[])message.getPayload())); adapter.stop(); } - - private void waitListening(AbstractInternetProtocolReceivingChannelAdapter adapter) throws Exception { - int n = 0; - while (!adapter.isListening()) { - Thread.sleep(100); - if (n++ > 100) { - throw new Exception("Gateway failed to listen"); - } - } - - } - - } 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 e3bf27e951..6b321254e7 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 @@ -38,6 +38,7 @@ import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.udp.DatagramPacketMessageMapper; import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; import org.springframework.integration.ip.udp.UnicastSendingMessageHandler; +import org.springframework.integration.ip.util.SocketUtils; import org.springframework.integration.message.MessageBuilder; /** @@ -47,9 +48,11 @@ import org.springframework.integration.message.MessageBuilder; */ public class DatagramPacketSendingHandlerTests { + private boolean noMulticast; + @Test public void verifySend() throws Exception { - final int testPort = 27816; + final int testPort = SocketUtils.findAvailableUdpSocket(); byte[] buffer = new byte[8]; final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length); final CountDownLatch latch = new CountDownLatch(1); @@ -83,11 +86,12 @@ public class DatagramPacketSendingHandlerTests { @Test public void verifySendWithAck() throws Exception { - final int testPort = 27816; - final int ackPort = 17816; + final int testPort = SocketUtils.findAvailableUdpSocket(); + final int ackPort = SocketUtils.findAvailableUdpSocket(testPort + 1); byte[] buffer = new byte[1000]; final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length); - final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); UnicastSendingMessageHandler handler = new UnicastSendingMessageHandler("localhost", testPort, true, true, "localhost", ackPort, 5000); @@ -95,6 +99,7 @@ public class DatagramPacketSendingHandlerTests { public void run() { try { DatagramSocket socket = new DatagramSocket(testPort); + latch1.countDown(); socket.receive(receivedPacket); socket.close(); DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); @@ -108,17 +113,17 @@ public class DatagramPacketSendingHandlerTests { DatagramSocket out = new DatagramSocket(); out.send(ackPack); out.close(); - latch.countDown(); + latch2.countDown(); } catch (Exception e) { e.printStackTrace(); } } }); - Thread.sleep(3000); + latch1.await(3000, TimeUnit.MILLISECONDS); String payload = "foobar"; handler.handleMessage(MessageBuilder.withPayload(payload).build()); - assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS)); byte[] src = receivedPacket.getData(); int length = receivedPacket.getLength(); int offset = receivedPacket.getOffset(); @@ -131,10 +136,11 @@ public class DatagramPacketSendingHandlerTests { @Test @Ignore public void verifySendMulticast() throws Exception { - final int testPort = 27816; + final int testPort = SocketUtils.findAvailableUdpSocket(); final String multicastAddress = "225.6.7.8"; final String payload = "foo"; - final CountDownLatch latch = new CountDownLatch(2); + final CountDownLatch latch1 = new CountDownLatch(2); + final CountDownLatch latch2 = new CountDownLatch(2); Runnable catcher = new Runnable() { public void run() { try { @@ -143,6 +149,7 @@ public class DatagramPacketSendingHandlerTests { MulticastSocket socket = new MulticastSocket(testPort); InetAddress group = InetAddress.getByName(multicastAddress); socket.joinGroup(group); + latch1.countDown(); LogFactory.getLog(getClass()) .debug(Thread.currentThread().getName() + " waiting for packet"); socket.receive(receivedPacket); @@ -155,9 +162,11 @@ public class DatagramPacketSendingHandlerTests { assertEquals(payload, new String(dest)); LogFactory.getLog(getClass()) .debug(Thread.currentThread().getName() + " received packet"); - latch.countDown(); + latch2.countDown(); } catch (Exception e) { + noMulticast = true; + latch1.countDown(); e.printStackTrace(); } } @@ -165,21 +174,25 @@ public class DatagramPacketSendingHandlerTests { Executor executor = Executors.newFixedThreadPool(2); executor.execute(catcher); executor.execute(catcher); - Thread.sleep(1000); + latch1.await(3000, TimeUnit.MILLISECONDS); + if (noMulticast) { + return; + } MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort); handler.handleMessage(MessageBuilder.withPayload(payload).build()); - assertTrue(latch.await(3000, TimeUnit.MILLISECONDS)); + assertTrue(latch2.await(3000, TimeUnit.MILLISECONDS)); handler.shutDown(); } @Test @Ignore public void verifySendMulticastWithAcks() throws Exception { - final int testPort = 27816; - final int ackPort = 17817; + final int testPort = SocketUtils.findAvailableUdpSocket(); + final int ackPort = SocketUtils.findAvailableUdpSocket(testPort + 1); final String multicastAddress = "225.6.7.8"; final String payload = "foobar"; - final CountDownLatch latch = new CountDownLatch(2); + final CountDownLatch latch1 = new CountDownLatch(2); + final CountDownLatch latch2 = new CountDownLatch(2); Runnable catcher = new Runnable() { public void run() { try { @@ -188,6 +201,7 @@ public class DatagramPacketSendingHandlerTests { MulticastSocket socket = new MulticastSocket(testPort); InetAddress group = InetAddress.getByName(multicastAddress); socket.joinGroup(group); + latch1.countDown(); LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " waiting for packet"); socket.receive(receivedPacket); socket.close(); @@ -209,9 +223,11 @@ public class DatagramPacketSendingHandlerTests { DatagramSocket out = new DatagramSocket(); out.send(ackPack); out.close(); - latch.countDown(); + latch2.countDown(); } catch (Exception e) { + noMulticast = true; + latch1.countDown(); e.printStackTrace(); } } @@ -219,13 +235,16 @@ public class DatagramPacketSendingHandlerTests { Executor executor = Executors.newFixedThreadPool(2); executor.execute(catcher); executor.execute(catcher); - Thread.sleep(3000); + latch1.await(3000, TimeUnit.MILLISECONDS); + if (noMulticast) { + return; + } MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort, true, - true, "localhost", ackPort, 500000);; + true, "localhost", ackPort, 500000); handler.setMinAcksForSuccess(2); handler.handleMessage(MessageBuilder.withPayload(payload).build()); - assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS)); handler.shutDown(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java index 8bd2a7b3c4..eee19fd958 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java @@ -58,10 +58,7 @@ public class MultiClientTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); - while (!adapter.isRunning()) { - Thread.sleep(50); // wait for server to start listening - } - Thread.sleep(250); // wait for listener + SocketUtils.waitListening(adapter); for (int i = 0; i < drivers; i++) { Thread t = new Thread( new Runnable() { public void run() { @@ -102,10 +99,7 @@ public class MultiClientTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); - while (!adapter.isRunning()) { - Thread.sleep(50); // wait for server to start listening - } - Thread.sleep(250); // wait for listener + SocketUtils.waitListening(adapter); for (int i = 0; i < drivers; i++) { final int j = i; Thread t = new Thread( new Runnable() { @@ -150,10 +144,7 @@ public class MultiClientTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); - while (!adapter.isRunning()) { - Thread.sleep(50); // wait for server to start listening - } - Thread.sleep(250); // wait for listener + SocketUtils.waitListening(adapter); for (int i = 0; i < drivers; i++) { final int j = i; Thread t = new Thread( new Runnable() { @@ -161,7 +152,7 @@ public class MultiClientTests { UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( "localhost", adapter.getPort(), true, true, "localhost", - SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000), + SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1100), 10000); while (true) { Message message = queueIn.receive(); 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 new file mode 100644 index 0000000000..fd2c6ae3dc --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -0,0 +1,133 @@ +package org.springframework.integration.ip.udp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.Inet4Address; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.NetworkInterface; +import java.util.Enumeration; + +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.util.SocketUtils; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + + +public class UdpChannelAdapterTests { + + @SuppressWarnings("unchecked") + @Test + public void testUnicastReceiver() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = SocketUtils.findAvailableUdpSocket(); + UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + SocketUtils.setLocalNicIfPossible(adapter); + adapter.start(); + SocketUtils.waitListening(adapter); + + Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); + DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); + DatagramPacket packet = mapper.fromMessage(message); + packet.setSocketAddress(new InetSocketAddress("localhost", port)); + new DatagramSocket(SocketUtils.findAvailableUdpSocket()).send(packet); + Message receivedMessage = (Message) channel.receive(2000); + assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testUnicastSender() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = SocketUtils.findAvailableUdpSocket(); + UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + SocketUtils.setLocalNicIfPossible(adapter); + adapter.start(); + SocketUtils.waitListening(adapter); + + String whichNic = SocketUtils.chooseANic(false); + UnicastSendingMessageHandler handler = new UnicastSendingMessageHandler( + "localhost", port, false, true, whichNic, + SocketUtils.findAvailableUdpSocket(), 500000); + handler.setLocalAddress(whichNic); + Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); + handler.handleMessage(message); + Message receivedMessage = (Message) channel.receive(2000); + assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testMulticastReceiver() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = SocketUtils.findAvailableUdpSocket(); + MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.8", port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + String nic = SocketUtils.chooseANic(true); + if (nic == null) { // no multicast support + LogFactory.getLog(this.getClass()).error("No Multicast support"); + return; + } + adapter.setLocalAddress(nic); + adapter.start(); + SocketUtils.waitListening(adapter); + + 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)); + new DatagramSocket(0, Inet4Address.getByName(nic)).send(packet); + + Message receivedMessage = (Message) channel.receive(2000); + assertNotNull(receivedMessage); + assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testMulticastSender() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = SocketUtils.findAvailableUdpSocket(); + UnicastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.9", port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + String nic = SocketUtils.chooseANic(true); + if (nic == null) { // no multicast support + LogFactory.getLog(this.getClass()).error("No Multicast support"); + return; + } + adapter.setLocalAddress(nic); + adapter.start(); + SocketUtils.waitListening(adapter); + + MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler("225.6.7.9", port); + handler.setLocalAddress(nic); + Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); + handler.handleMessage(message); + + Message receivedMessage = (Message) channel.receive(2000); + assertNotNull(receivedMessage); + assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); + } + + +} 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 7540264faf..26ed4a3f5f 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 @@ -34,6 +34,7 @@ import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessagingException; import org.springframework.integration.message.StringMessage; /** @@ -97,6 +98,10 @@ public class UdpMulticastEndToEndTests implements Runnable { catch (InterruptedException e) { e.printStackTrace(); } + } catch (MessagingException e) { + // no multicast this host + e.printStackTrace(); + return; } finally { if (hangAroundFor == 0) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java index 2dc18e8e0b..ba9306c3ad 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java @@ -19,15 +19,19 @@ import java.io.ObjectOutputStream; import java.io.OutputStream; import java.net.DatagramSocket; import java.net.InetAddress; +import java.net.NetworkInterface; import java.net.ServerSocket; import java.net.Socket; +import java.net.UnknownHostException; import java.nio.ByteBuffer; +import java.util.Enumeration; import java.util.concurrent.CountDownLatch; import javax.net.ServerSocketFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; /** * TCP/IP Test utilities. @@ -329,4 +333,43 @@ public class SocketUtils { return findAvailableUdpSocket(9876); } + public static void setLocalNicIfPossible( + AbstractInternetProtocolReceivingChannelAdapter adapter) + throws UnknownHostException { + InetAddress[] nics = InetAddress.getAllByName(null); + if (nics.length > 0) { + // just listen on the loopback interface + String loopBack = nics[0].getHostAddress(); + adapter.setLocalAddress(loopBack); + } + } + + 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; + } + return null; + } + + public static void waitListening(AbstractInternetProtocolReceivingChannelAdapter adapter) throws Exception { + int n = 0; + while (!adapter.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + throw new Exception("Gateway failed to listen"); + } + } + + } + + + }