initial commit of IP code

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java

@@ -0,0 +1,86 @@
+/*
+ * 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.util.Date;
+
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for inbound TCP/UDP Channel Adapters.
+ * 
+ * @author Mark Fisher
+ * @author Gary Russell
+ * @since 2.0
+ */
+public abstract class AbstractInternetProtocolReceivingChannelAdapter
+		extends MessageProducerSupport implements Runnable, CommonSocketOptions {
+
+	protected final int port;
+
+	protected volatile int soTimeout = 60 * 1000;
+
+	protected volatile int soReceiveBufferSize = -1;
+
+	protected volatile int soSendBufferSize = -1;
+
+	protected volatile int receiveBufferSize = 2048;
+
+	protected volatile boolean active;
+
+
+	public AbstractInternetProtocolReceivingChannelAdapter(int port) {
+		this.port = port;
+	}
+
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.SocketOptions#setSoTimeout(int)
+	 */
+	public void setSoTimeout(int soTimeout) {
+		this.soTimeout = soTimeout;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.SocketOptions#setSoReceiveBufferSize(int)
+	 */
+	public void setSoReceiveBufferSize(int soReceiveBufferSize) {
+		this.soReceiveBufferSize = soReceiveBufferSize;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.SocketOptions#setSoSendBufferSize(int)
+	 */
+	public void setSoSendBufferSize(int soSendBufferSize) {
+		this.soSendBufferSize = soSendBufferSize;
+	}
+
+	public void setReceiveBufferSize(int receiveBufferSize) {
+		this.receiveBufferSize = receiveBufferSize;
+	}
+
+	@Override
+	protected void doStart() {
+		TaskScheduler taskScheduler = this.getTaskScheduler();
+		Assert.state(taskScheduler != null, "taskScheduler is required");
+		this.active = true;
+		taskScheduler.schedule(this, new Date());
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java

@@ -0,0 +1,76 @@
+/*
+ * 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.InetSocketAddress;
+import java.net.SocketAddress;
+
+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 int soReceiveBufferSize = -1;
+
+	protected int soSendBufferSize = -1;
+
+	protected int soTimeout = -1;
+
+
+	public AbstractInternetProtocolSendingMessageHandler(String host, int port) {
+		Assert.notNull(host, "host must not be null");
+		this.destinationAddress = new InetSocketAddress(host, port);
+	}
+
+
+	/**
+	 * @see {@link Socket#setSoTimeout(int)} and {@link DatagramSocket#setSoTimeout(int)}
+	 * @param timeout
+	 */
+	public void setSoTimeout(int timeout) {
+		this.soTimeout = timeout;
+	}
+
+	/**
+	 * @see {@link Socket#setReceiveBufferSize(int)} and {@link DatagramSocket#setReceiveBufferSize(int)}
+	 * @param size
+	 */
+	public void setSoReceiveBufferSize(int size) {
+		this.soReceiveBufferSize = size;
+	}
+
+	/**
+	 * @see {@link Socket#setSendBufferSize(int)} and {@link DatagramSocket#setSendBufferSize(int)}
+	 * @param size
+	 */
+	public void setSoSendBufferSize(int size) {
+		this.soSendBufferSize = size;
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public interface CommonSocketOptions {
+
+	void setSoTimeout(int soTimeout);
+
+	void setSoReceiveBufferSize(int soReceiveBufferSize);
+
+	void setSoSendBufferSize(int soSendBufferSize);
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java

@@ -0,0 +1,44 @@
+/*
+ * 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 org.springframework.integration.core.MessageHeaders;
+
+/**
+ * Headers for Messages mapped from UDP datagram packets.
+ * 
+ * @author Mark Fisher
+ * @author Gary Russell
+ * @since 2.0
+ */
+public abstract class IpHeaders {
+
+	private static final String PREFIX = MessageHeaders.PREFIX;
+
+	private static final String IP = "ip_";
+
+	private static final String TCP = "tcp_";
+
+	private static final String UDP = "udp_";
+
+	public static final String HOSTNAME = PREFIX + IP + "hostname";
+
+	public static final String IP_ADDRESS = PREFIX + IP + "address"; 
+
+	public static final String ACK_ADDRESS = PREFIX + "ackTo";
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -0,0 +1,124 @@
+/*
+ * 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Utility methods and constants for IP adapter parsers.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public abstract class IpAdapterParserUtils {
+
+	static final String IP_PROTOCOL_ATTRIBUTE = "protocol";
+
+	static final String UDP_MULTICAST = "multicast";
+
+	static final String MULTICAST_ADDRESS = "multicast-address";
+
+	static final String PORT = "port";
+
+	static final String HOST = "host";
+
+	static final String CHECK_LENGTH = "check-length";
+
+	static final String SO_TIMEOUT = "so-timeout";
+
+	static final String SO_RECEIVE_BUFFER_SIZE = "so-receive-bufffer-size";
+
+	static final String SO_SEND_BUFFER_SIZE = "so-send-buffer-size";
+
+	static final String RECEIVE_BUFFER_SIZE = "receive-buffer-size";
+
+	static final String POOL_SIZE = "pool-size";
+
+	static final String ACK = "acknowledge";
+
+	static final String ACK_HOST = "ack-host";
+
+	static final String ACK_PORT = "ack-port";
+
+	static final String ACK_TIMEOUT = "ack-timeout";
+
+	static final String MIN_ACKS_SUCCESS = "min-acks-for-success";
+
+	static final String TIME_TO_LIVE = "time-to-live";
+
+
+	/**
+	 * Adds a constructor-arg to the bean definition with the value
+	 * of the attribute whose name is provided if that attribute is
+	 * defined in the given element.
+	 * 
+	 * @param beanDefinition the bean definition to be configured
+	 * @param element the XML element where the attribute should be defined
+	 * @param attributeName the name of the attribute whose value will be
+	 * used to populate the property
+	 */
+	public static void addConstuctirValueIfAttributeDefined(BeanDefinitionBuilder builder,
+			Element element, String attributeName, boolean trueFalse) {
+		String attributeValue = element.getAttribute(attributeName);
+		if (StringUtils.hasText(attributeValue)) {
+			builder.addConstructorArgValue(attributeValue);
+		}
+	}
+
+	static String getProtocol(Element element) {
+		String protocol = element.getAttribute(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE);
+		if (!StringUtils.hasText(protocol)) {
+			throw new BeanCreationException(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE + 
+					" is required for an IP channel adapter");
+		}
+		protocol = protocol.trim();
+		if (!protocol.equals("tcp") && !protocol.equals("udp")) {
+			throw new BeanCreationException(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE + 
+					" must be 'tcp' or 'udp' for an IP channel adapter");
+		}
+		return protocol;
+	}
+
+	static String getPort(Element element) {
+		String port = element.getAttribute(IpAdapterParserUtils.PORT);
+		if (!StringUtils.hasText(port)) {
+			throw new BeanCreationException(IpAdapterParserUtils.PORT +
+					" is required for IP channel adapters");
+		}
+		return port;
+	}
+
+	static String getMulticast(Element element) {
+		String multicast = element.getAttribute(IpAdapterParserUtils.UDP_MULTICAST);
+		if (!StringUtils.hasText(multicast)) {
+			multicast = "false";
+		}
+		return multicast;
+	}
+
+	static void addCommonSocketOptions(BeanDefinitionBuilder builder, Element element) {
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_TIMEOUT);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_RECEIVE_BUFFER_SIZE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_SEND_BUFFER_SIZE);
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java

@@ -0,0 +1,78 @@
+/*
+ * 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
+import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
+import org.springframework.util.StringUtils;
+
+/**
+ * Channel Adapter that receives UDP datagram packets and maps them to Messages.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser {
+
+	protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
+		String protocol = IpAdapterParserUtils.getProtocol(element);
+		String multicast = IpAdapterParserUtils.getMulticast(element);
+		BeanDefinitionBuilder builder = null;
+		if (protocol.equals("tcp")) {
+			throw new BeanCreationException("tcp not yet supported");
+		}
+		else if (protocol.equals("udp")) {
+			if (multicast.equals("false")) {
+				builder = BeanDefinitionBuilder
+						.genericBeanDefinition(UnicastReceivingChannelAdapter.class);
+			}
+			else {
+				builder = BeanDefinitionBuilder
+						.genericBeanDefinition(MulticastReceivingChannelAdapter.class);
+				String mcAddress = element
+						.getAttribute(IpAdapterParserUtils.MULTICAST_ADDRESS);
+				if (!StringUtils.hasText(mcAddress)) {
+					throw new BeanCreationException(
+							IpAdapterParserUtils.MULTICAST_ADDRESS
+									+ " is required for a multicast UDP/IP channel adapter");
+				}
+				builder.addConstructorArgValue(mcAddress);
+			}
+		}
+		String port = IpAdapterParserUtils.getPort(element);
+		builder.addConstructorArgValue(port);
+		IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
+				element, IpAdapterParserUtils.CHECK_LENGTH, true);
+		IpAdapterParserUtils.addCommonSocketOptions(builder, element);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.POOL_SIZE);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder,
+				element, "channel", "outputChannel");
+		return builder.getBeanDefinition();
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java

@@ -0,0 +1,34 @@
+/*
+ * 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.config;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * Namespace handler for Spring Integration's <em>ip</em> namespace.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	public void init() {
+		this.registerBeanDefinitionParser("inbound-channel-adapter", new IpInboundChannelAdapterParser());
+		this.registerBeanDefinitionParser("outbound-channel-adapter", new IpOutboundChannelAdapterParser());
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java

@@ -0,0 +1,100 @@
+/*
+ * 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
+import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+	protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+		String protocol = IpAdapterParserUtils.getProtocol(element);
+		BeanDefinitionBuilder builder = null;
+		if (protocol.equals("tcp")) {
+			throw new BeanCreationException("tcp not yet supported");
+		}
+		else if (protocol.equals("udp")) {
+			String multicast = IpAdapterParserUtils.getMulticast(element);
+			if (multicast.equals("true")) {
+				builder = BeanDefinitionBuilder
+						.genericBeanDefinition(MulticastSendingMessageHandler.class);
+				IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
+						element, IpAdapterParserUtils.MIN_ACKS_SUCCESS,
+						"minAcksForSuccess");
+				IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
+						element, IpAdapterParserUtils.TIME_TO_LIVE,
+						"timeToLive");
+			}
+			else {
+				builder = BeanDefinitionBuilder
+						.genericBeanDefinition(UnicastSendingMessageHandler.class);
+			}
+		}
+		String host = element.getAttribute(IpAdapterParserUtils.HOST);
+		if (!StringUtils.hasText(host)) {
+			throw new BeanCreationException(IpAdapterParserUtils.HOST
+					+ " is required for IP outbound channel adapters");
+		}
+		builder.addConstructorArgValue(host);
+		String port = IpAdapterParserUtils.getPort(element);
+		builder.addConstructorArgValue(port);
+		IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
+				element, IpAdapterParserUtils.CHECK_LENGTH, true);
+		IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
+				element, IpAdapterParserUtils.ACK, true);
+		IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
+				element, IpAdapterParserUtils.ACK_HOST, false);
+		IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
+				element, IpAdapterParserUtils.ACK_PORT, false);
+		IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
+				element, IpAdapterParserUtils.ACK_TIMEOUT, false);
+		String ack = element.getAttribute(IpAdapterParserUtils.ACK);
+		if (ack.equals("true")) {
+			if (!StringUtils.hasText(element
+					.getAttribute(IpAdapterParserUtils.ACK_HOST))
+					|| !StringUtils.hasText(element
+							.getAttribute(IpAdapterParserUtils.ACK_PORT))
+					|| !StringUtils.hasText(element
+							.getAttribute(IpAdapterParserUtils.ACK_TIMEOUT))) {
+				throw new BeanCreationException("When "
+						+ IpAdapterParserUtils.ACK + " is true, "
+						+ IpAdapterParserUtils.ACK_HOST + ", "
+						+ IpAdapterParserUtils.ACK_PORT + ", and "
+						+ IpAdapterParserUtils.ACK_TIMEOUT
+						+ " must be supplied");
+			}
+		}
+		IpAdapterParserUtils.addCommonSocketOptions(builder, element);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
+		return builder.getBeanDefinition();
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java

@@ -0,0 +1,240 @@
+/*
+ * 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.UnsupportedEncodingException;
+import java.net.DatagramPacket;
+import java.nio.ByteBuffer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.springframework.integration.adapter.MessageMappingException;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageHeaders;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.message.InboundMessageMapper;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.OutboundMessageMapper;
+import org.springframework.util.Assert;
+
+/**
+ * Message Mapper for converting to and from UDP DatagramPackets. When
+ * converting to a Message, the payload will be a byte array containing the
+ * data from the received packet. When converting from a Message, the payload
+ * may be either a byte array or a String. The default charset for converting
+ * a String to a byte array is UTF-8, but that may be changed by invoking the
+ * {@link #setCharset(String)} method.
+ * 
+ * By default, the UDP messages will be unreliable (truncation may occur on
+ * the receiving end; packets may be lost).
+ * 
+ * Reliability can be enhanced by one or both of the following techniques:
+ * <ul>
+ *   <li>including a binary message length at the beginning of the packet</li>
+ *   <li>requesting a receipt acknowledgment</li>
+ * </ul>
+ * 
+ * @author Mark Fisher
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>,
+		OutboundMessageMapper<DatagramPacket> {
+
+	private volatile String charset = "UTF-8";
+
+	private boolean acknowledge = false;
+
+	private String ackAddress;
+	
+	private boolean lengthCheck = false;
+
+	private static Pattern udpHeadersPattern = 
+		Pattern.compile(IpHeaders.ACK_ADDRESS + "=" + "([^;]*);\\" + 
+				MessageHeaders.ID + "=" + "([^;]*);");
+
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public void setAcknowledge(boolean acknowledge) {
+		this.acknowledge = acknowledge;
+	}
+
+	public void setAckAddress(String ackAddress) {
+		this.ackAddress = ackAddress;
+	}
+
+	public void setLengthCheck(boolean lengthCheck) {
+		this.lengthCheck = lengthCheck;
+	}
+
+	/**
+	 * Raw byte[] from message, possibly with a length field up front.
+	 */
+	public DatagramPacket fromMessage(Message<?> message) throws Exception {
+		if (this.acknowledge) {
+			return fromMessageWithAck(message);
+		}
+		byte[] bytes = getPayloadAsBytes(message);
+		if (this.lengthCheck) {
+			ByteBuffer buffer = ByteBuffer.allocate(bytes.length + 4);
+			// insert the length (not including the length bytes)
+			// default ByteOrder is	ByteOrder.BIG_ENDIAN (network byte order)
+			buffer.putInt(bytes.length);
+			buffer.put(bytes);
+			bytes = buffer.array();
+		}
+		return new DatagramPacket(bytes, bytes.length);
+	}
+
+	/**
+	 * Prefix raw byte[] from message with 'acknowledge to' and 'message id' "headers".
+	 * @param message
+	 * @param ackTo
+	 * @return
+	 * @throws Exception
+	 */
+	private DatagramPacket fromMessageWithAck(Message<?> message) throws Exception {
+		Assert.hasLength(this.ackAddress);
+		byte[] bytes = getPayloadAsBytes(message);
+		ByteBuffer buffer = ByteBuffer.allocate(100 + bytes.length);
+		if (this.lengthCheck) {
+			buffer.putInt(0); // placeholder for length
+		}
+		buffer.put(IpHeaders.ACK_ADDRESS.getBytes(this.charset));
+		buffer.put((byte) '=');
+		buffer.put(this.ackAddress.getBytes(this.charset));
+		buffer.put((byte) ';');		
+		buffer.put(MessageHeaders.ID.getBytes(this.charset));
+		buffer.put((byte) '=');
+		buffer.put(message.getHeaders().getId().toString().getBytes(this.charset));
+		buffer.put((byte) ';');
+		int headersLength = buffer.position() - 4;
+		buffer.put(bytes);
+		if (this.lengthCheck) {
+			// insert the length (not including the length bytes)
+			// default ByteOrder is	ByteOrder.BIG_ENDIAN (network byte order)
+			buffer.putInt(0, bytes.length + headersLength);
+		}
+		return new DatagramPacket(buffer.array(), buffer.position());
+	}
+
+	private byte[] getPayloadAsBytes(Message<?> message) {
+		byte[] bytes = null;
+		Object payload = message.getPayload();
+		if (payload instanceof byte[]) {
+			bytes = (byte[]) payload;
+		}
+		else if (payload instanceof String) {
+			try {
+				bytes = ((String) payload).getBytes(this.charset);
+			}
+			catch (UnsupportedEncodingException e) {
+				throw new MessageHandlingException(message, e);
+			}
+		}
+		else {
+			throw new MessageHandlingException(message, "The datagram packet mapper expects " +
+					"either a byte array or String payload, but received: " + payload.getClass());
+		}
+		return bytes;
+	}
+
+	public Message<byte[]> toMessage(DatagramPacket packet) throws Exception {
+		int offset = packet.getOffset();
+		int length = packet.getLength();
+		byte[] payload;
+		ByteBuffer buffer = ByteBuffer.wrap(packet.getData(), offset, length);
+		Message<byte[]> message = null;
+		if (this.lengthCheck) {
+			int declaredLength = buffer.getInt();
+			if (declaredLength != (length - 4)) {
+				throw new MessageMappingException("Incorrect length; expected " + (declaredLength + 4) + ", received " + length);
+			}
+			offset += 4;
+			length -= 4;
+		}
+		// Peek at the message in case they didn't configure us for ack but the sending
+		// side expects it.
+		if (this.acknowledge || startsWith(buffer, IpHeaders.ACK_ADDRESS)) {
+			try {
+				String headers = new String(packet.getData(), offset, length, this.charset);
+				Matcher matcher = udpHeadersPattern.matcher(headers);
+				if (matcher.find()) {
+					// Strip off the ack headers and put in Message headers
+					length = length - matcher.end();
+					payload = new byte[length];
+					System.arraycopy(packet.getData(), offset + matcher.end(), payload, 0, length);
+					message = MessageBuilder.withPayload(payload)
+							.setHeader(MessageHeaders.ID, matcher.group(2))
+							.setHeader(IpHeaders.ACK_ADDRESS, matcher.group(1))
+							.setHeader(IpHeaders.HOSTNAME, packet.getAddress().getHostName())
+							.setHeader(IpHeaders.IP_ADDRESS, packet.getAddress().getHostAddress())
+							.build();
+				}  // on no match, just treat as simple payload
+			}
+			catch (UnsupportedEncodingException e) {
+				throw new MessageMappingException("Invalid charset", e);
+			}
+		}
+		if (message == null) {
+			payload = new byte[length];
+			System.arraycopy(packet.getData(), offset, payload, 0, length);
+			if (payload.length > 0) {
+				message = MessageBuilder.withPayload(payload)
+						.setHeader(IpHeaders.HOSTNAME, packet.getAddress().getHostName())
+						.setHeader(IpHeaders.IP_ADDRESS, packet.getAddress().getHostAddress())
+						.build();
+			}
+		}
+		return message;
+	}
+
+	/**
+	 * Peeks at data in he buffer to see if starts with the prefix.
+	 * @param buffer
+	 * @param prefix
+	 * @return
+	 */
+	private boolean startsWith(ByteBuffer buffer, String prefix) {
+		int pos = buffer.position();
+		if (buffer.limit() - pos < prefix.length()) {
+			return false;
+		}
+		try {
+			byte[] comparing;
+			comparing = prefix.getBytes(this.charset);
+			for (int i = 0; i < comparing.length; i++) {
+				if (buffer.get() != comparing[i]) {
+					return false;
+				}
+			}
+			return true;
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new MessageMappingException("Invalid charset", e);
+		}
+		finally {
+			//reposition the buffer
+			buffer.position(pos);
+		}
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java

@@ -0,0 +1,72 @@
+/*
+ * 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.MulticastSocket;
+
+import org.springframework.integration.core.MessagingException;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAdapter {
+
+	protected String group;
+
+
+	/**
+	 * @param port
+	 */
+	public MulticastReceivingChannelAdapter(String group, int port) {
+		super(port);
+		this.group = group;
+	}
+
+	/**
+	 * @param port
+	 * @param lengthCheck
+	 */
+	public MulticastReceivingChannelAdapter(String group, int port, boolean lengthCheck) {
+		super(port, lengthCheck);
+		this.group = group;
+	}
+
+
+	@Override
+	protected synchronized DatagramSocket getSocket() {
+		if (this.socket == null) {
+			try {
+				MulticastSocket socket = new MulticastSocket(this.port);
+				socket.setSoTimeout(this.soTimeout);
+				if (this.soReceiveBufferSize > 0) {
+					socket.setReceiveBufferSize(this.soReceiveBufferSize);
+				}
+				socket.joinGroup(InetAddress.getByName(this.group));
+				this.socket = socket;
+			}
+			catch (IOException e) {
+				throw new MessagingException("failed to create DatagramSocket", e);
+			}
+		}
+		return this.socket;
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java

@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+
+	public MulticastSendingMessageHandler(String address, int port) {
+		super(address, port);
+	}
+
+	public MulticastSendingMessageHandler(String address, int port, boolean lengthCheck) {
+		super(address, port, lengthCheck);
+	}
+
+	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;
+	}
+
+	public void setTimeToLive(int timeToLive) {
+		this.timeToLive = timeToLive;
+	}
+
+	protected 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;
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UdpSocket.java

@@ -0,0 +1,80 @@
+/*
+ * 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.net.DatagramSocket;
+import java.net.MulticastSocket;
+import java.net.SocketException;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class UdpSocket {
+
+	private DatagramSocket datagramSocket;
+
+	private MulticastSocket multicastSocket;
+
+
+	/**
+	 * Creates a Unicast UdpSocket on the specified port.
+	 * @param port The port.
+	 */
+	public UdpSocket(int port) {
+		// TODO
+	}
+
+	/**
+	 * Creates a multicast UdpSocket on the specified port which will join
+	 * the specified group (multicast ip address).
+	 * @param group The group (multicast ip address) to join.
+	 * @param port The port.
+	 */
+	public UdpSocket(String group, int port) {
+		// TODO
+	}
+
+
+	public void setReceiveBufferSize(int size) throws SocketException {
+		if (datagramSocket != null) {
+			datagramSocket.setReceiveBufferSize(size);
+		}
+		if (multicastSocket != null) {
+			multicastSocket.setReceiveBufferSize(size);
+		}
+	}
+
+	public void setSoTimeout(int timeout) throws SocketException {
+		if (datagramSocket != null) {
+			datagramSocket.setSoTimeout(timeout);
+		}
+		if (multicastSocket != null) {
+			multicastSocket.setSoTimeout(timeout);
+		}
+	}
+
+	public void close() {
+		if (datagramSocket != null) {
+			datagramSocket.close();
+		}
+		if (multicastSocket != null) {
+			multicastSocket.close();
+		}
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java

@@ -0,0 +1,193 @@
+/*
+ * 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.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.util.concurrent.ThreadFactory;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageHeaders;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolReceivingChannelAdapter {
+
+	protected volatile DatagramSocket socket;
+
+	protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
+
+	protected volatile ThreadPoolTaskScheduler threadPoolTaskScheduler;
+
+	protected volatile int poolSize = -1;
+
+	private static Pattern addressPattern = Pattern.compile("([^:]*):([0-9]*)");
+
+
+	public UnicastReceivingChannelAdapter(int port) {
+		super(port);
+		mapper.setLengthCheck(false);
+	}
+
+	public UnicastReceivingChannelAdapter(int port, boolean lengthCheck) {
+		super(port);
+		mapper.setLengthCheck(lengthCheck);
+	}
+
+
+	public void setPoolSize(int poolSize) {
+		this.poolSize = poolSize;
+	}
+
+	public void run() {
+		if (logger.isDebugEnabled()) {
+			logger.debug("UDP Receiver running...");
+		}
+		if (this.active && this.threadPoolTaskScheduler == null) {
+			this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
+			this.threadPoolTaskScheduler.setThreadFactory(new ThreadFactory() {
+				public Thread newThread(Runnable runner) {
+					Thread thread = new Thread(runner);
+					thread.setName("UDP-Incoming-Msg-Handler");
+					thread.setDaemon(true);
+					return thread;
+				}
+			});
+			if (this.poolSize > 0) {
+				this.threadPoolTaskScheduler.setPoolSize(this.poolSize);
+			}
+			this.threadPoolTaskScheduler.initialize();
+		}
+
+		// Do as little as possible here so we can loop around and catch the next packet.
+		// Just schedule the packet for processing.
+		while (this.active) {
+			try {
+				scheduleSendMessage(receive());
+			}
+			catch (SocketTimeoutException e) {
+				// continue
+			}
+			catch (SocketException e) {
+				doStop(); 
+			}
+			catch (Exception e) {
+				if (e instanceof MessagingException) {
+					throw (MessagingException) e;
+				}
+				throw new MessagingException("failed to receive DatagramPacket", e);
+			}
+		}
+	}
+
+	private void sendAck(Message<byte[]> message) {
+		MessageHeaders headers = message.getHeaders();
+		Object id = headers.getId();
+		byte[] ack = id.toString().getBytes();
+		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);
+		}
+		String host = mat.group(1);
+		int port = Integer.parseInt(mat.group(2));
+		InetSocketAddress whereTo = new InetSocketAddress(host, port);
+		if (logger.isDebugEnabled()) {
+			logger.debug("Sending ack for " + id + " to " + ackAddress);
+		}
+		try {
+			DatagramPacket ackPack = new DatagramPacket(ack, ack.length, whereTo);
+			DatagramSocket out = new DatagramSocket();
+			if (this.soSendBufferSize > 0) {
+				out.setSendBufferSize(this.soSendBufferSize);
+			}
+			out.send(ackPack);
+			out.close();
+		}
+		catch (IOException e) {
+			throw new MessagingException(message, "Failed to send acknowledgment", e);
+		}
+	}
+
+	protected boolean scheduleSendMessage(final DatagramPacket packet) {
+		this.threadPoolTaskScheduler.execute(new Runnable(){
+			public void run() {
+				Message<byte[]> message = null;
+				try {
+					message = mapper.toMessage(packet);
+				}
+				catch (Exception e) {
+					logger.error("Failed to map packet to message ", e);
+				}
+				if (message != null) {
+					if (message.getHeaders().containsKey(IpHeaders.ACK_ADDRESS)) {
+						sendAck(message);
+					}
+					sendMessage(message);
+				}
+			}});
+		return true;
+	}
+
+	public DatagramPacket receive() throws Exception {
+		DatagramSocket socket = this.getSocket();
+		final byte[] buffer = new byte[this.receiveBufferSize];
+		DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
+		socket.receive(packet);
+		return packet;
+	}
+
+	protected synchronized DatagramSocket getSocket() {
+		if (this.socket == null) {
+			try {
+				this.socket = new DatagramSocket(this.port);
+				this.socket.setSoTimeout(this.soTimeout);
+				if (this.soReceiveBufferSize > 0) {
+					this.socket.setReceiveBufferSize(this.soReceiveBufferSize);
+				}
+			}
+			catch (SocketException e) {
+				throw new MessagingException("failed to create DatagramSocket", e);
+			}
+		}
+		return this.socket;
+	}
+
+	@Override
+	protected void doStop() {
+		this.active = false;
+		try {
+			this.socket.close();
+		}
+		catch (Exception e) {
+			// ignore
+		}
+	}
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java

@@ -0,0 +1,298 @@
+/*
+ * 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.BindException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
+import org.springframework.integration.message.MessageDeliveryException;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.MessageRejectedException;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageHandler} implementation that maps a Message into
+ * a UDP datagram packet and sends that to the specified host and port.
+ * 
+ * Messages can be basic, with no support for reliability, can be prefixed
+ * by a length so the receiving end can detect truncation, and can require 
+ * a UDP acknowledgment to confirm delivery.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class UnicastSendingMessageHandler extends
+		AbstractInternetProtocolSendingMessageHandler implements Runnable {
+
+	private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
+
+	protected volatile DatagramSocket socket;
+
+
+	/**
+	 * If true adds headers to instruct receiving adapter to return an ack.
+	 */
+	protected volatile boolean waitForAck = false;
+
+	protected volatile int ackPort;
+
+	protected volatile int ackTimeout = 5000;
+
+	protected volatile int ackCounter = 1;
+
+	protected volatile Map<String, CountDownLatch> ackControl = Collections
+			.synchronizedMap(new HashMap<String, CountDownLatch>());
+
+	protected volatile DatagramSocket ackSocket;
+
+	protected volatile ExecutorService executorService;
+
+	protected volatile Exception fatalException;
+
+
+	/**
+	 * Basic constructor; no reliability; no acknowledgment.
+	 * @param host Destination host.
+	 * @param port Destination port.
+	 */
+	public UnicastSendingMessageHandler(String host, int port) {
+		super(host, port);
+		this.mapper.setLengthCheck(false);
+		this.mapper.setAcknowledge(false);
+	}
+
+	/**
+	 * Can used to add a length to each packet which can be checked at the destination.
+	 * @param host Destination Host.
+	 * @param port Destination Port.
+	 * @param lengthCheck If true, packets will contain a length.
+	 */
+	public UnicastSendingMessageHandler(String host, int port, boolean lengthCheck) {
+		super(host, port);
+		this.mapper.setLengthCheck(lengthCheck);
+		this.mapper.setAcknowledge(false);
+	}
+
+	/**
+	 * Add an acknowledgment request to packets. 
+	 * @param host Destination Host.
+	 * @param port Destination Port.
+	 * @param acknowledge If true, packets will request acknowledgment.
+	 * @param ackHost The host to which acks should be sent. Required if ack true.
+	 * @param ackPort The port to which acks should be sent. 
+	 * @param ackTimeout How long we will wait (milliseconds) for the ack.
+	 */
+	public UnicastSendingMessageHandler(String host, 
+			                            int port, 
+			                            boolean acknowledge, 
+			                            String ackHost,
+			                            int ackPort,
+			                            int ackTimeout) {
+		super(host, port);
+		setReliabilityAttributes(false, acknowledge, ackHost, ackPort,
+				ackTimeout);
+	}
+
+	/**
+	 * Add a length and/or acknowledgment request to packets. 
+	 * @param host Destination Host.
+	 * @param port Destination Port.
+	 * @param lengthCheck If true, packets will contain a length.
+	 * @param acknowledge If true, packets will request acknowledgment.
+	 * @param ackHost The host to which acks should be sent. Required if ack true.
+	 * @param ackPort The port to which acks should be sent. 
+	 * @param ackTimeout How long we will wait (milliseconds) for the ack.
+	 */
+	public UnicastSendingMessageHandler(String host, 
+			                            int port, 
+			                            boolean lengthCheck, 
+			                            boolean acknowledge, 
+			                            String ackHost,
+			                            int ackPort,
+			                            int ackTimeout) {
+		super(host, port);
+		setReliabilityAttributes(lengthCheck, acknowledge, ackHost, ackPort,
+				ackTimeout);
+	}
+
+	protected void setReliabilityAttributes(boolean lengthCheck,
+			boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
+		this.mapper.setLengthCheck(lengthCheck);
+		this.waitForAck = acknowledge;
+		this.mapper.setAcknowledge(acknowledge);
+		this.mapper.setAckAddress(ackHost + ":" + ackPort);
+		this.ackPort = ackPort;
+		if (ackTimeout > 0) {
+			this.ackTimeout = ackTimeout;
+		}
+		if (acknowledge) {
+			Assert.hasLength(ackHost);
+			this.executorService = Executors
+					.newSingleThreadExecutor(new ThreadFactory() {
+						public Thread newThread(Runnable runner) {
+							Thread thread = new Thread(runner);
+							thread.setName("UDP-Ack-Handler");
+							thread.setDaemon(true);
+							return thread;
+						}
+					});
+			this.executorService.execute(this);
+		}
+	}
+
+	public void handleMessage(Message<?> message)
+			throws MessageRejectedException, MessageHandlingException,
+			MessageDeliveryException {
+		CountDownLatch countdownLatch = null;
+		String messageId = message.getHeaders().getId().toString();
+		try {
+			DatagramPacket packet;
+			if (this.waitForAck) {
+				if (this.fatalException != null) {
+					throw new MessagingException(message, "Acknowledgment failure", fatalException);
+				}
+				countdownLatch = new CountDownLatch(ackCounter);
+				this.ackControl.put(messageId, countdownLatch);
+			}
+			packet = this.mapper.fromMessage(message);
+			this.send(packet);
+			logger.debug("Sent packet for message id " + message.getHeaders().getId());
+			if (this.waitForAck) {
+				if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) {
+					throw new MessagingException(message, "Failed to received UDP Ack in " + ackTimeout + " millis");
+				}
+			}
+		}
+		catch (MessagingException e) {
+			throw e;
+		}
+		catch (Exception e) {
+			try{
+				socket.close();
+			}
+			catch (Exception e1) { }
+			socket = null;
+			throw new MessageHandlingException(message, "failed to send UDP packet", e);
+		} 
+		finally {
+			if (countdownLatch != null)
+				this.ackControl.remove(messageId);
+		}
+	}
+
+	protected void send(DatagramPacket packet) throws Exception {
+		DatagramSocket socket = this.getSocket();
+		packet.setSocketAddress(this.destinationAddress);
+		socket.send(packet);
+	}
+
+	protected DatagramSocket getSocket() throws IOException {
+		if (this.socket == null) {
+			this.socket = new DatagramSocket();
+			setSocketAttributes(this.socket);
+		}
+		return this.socket;
+	}
+
+	protected void setSocketAttributes(DatagramSocket socket) throws SocketException {
+		if (this.soTimeout >= 0) {
+			socket.setSoTimeout(this.soTimeout);
+		}
+		if (this.soSendBufferSize > 0) {
+			socket.setSendBufferSize(this.soSendBufferSize);
+		}
+	}
+
+	/**
+	 * Process acknowledgments, if requested.
+	 */
+	public void run() {
+		Exception fatalException = null;
+		try {
+			this.ackSocket = new DatagramSocket(this.ackPort);
+			if (this.soReceiveBufferSize > 0) {
+				ackSocket.setReceiveBufferSize(this.soReceiveBufferSize);
+			}
+			DatagramPacket ackPack = new DatagramPacket(new byte[100], 100);
+			while(true) {
+				this.ackSocket.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());
+				}
+				CountDownLatch latch = this.ackControl.get(id);
+				if (latch != null) {
+					latch.countDown();
+				}
+			}
+		}
+		catch (IOException e) {
+			if (this.ackSocket != null) {
+				logger.error("Error on UDP Acknowledge thread");
+				fatalException = e;
+			}
+		}
+		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);
+			}
+		}
+	}
+
+	/**
+	 * If exposed as an MBean, can be used to restart the ack thread if a fatal 
+	 * (bind) error occurred, without bouncing the JVM.
+	 */
+	public void restartAckThread() {
+		if (fatalException == null) {
+			return;
+		}
+		this.fatalException = null;
+		this.executorService.execute(this);
+	}
+
+	public void shutDown() {
+		DatagramSocket socket = this.ackSocket;
+		this.ackSocket = null;
+		if (socket != null) {
+			socket.close();
+		}
+	}
+
+}

org.springframework.integration.ip/src/main/resources/META-INF/spring.handlers

@@ -0,0 +1 @@
+http\://www.springframework.org/schema/integration/ip=org.springframework.integration.ip.config.IpNamespaceHandler

org.springframework.integration.ip/src/main/resources/META-INF/spring.schemas

@@ -0,0 +1,2 @@
+http\://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd=org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
+http\://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd=org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd

org.springframework.integration.ip/src/main/resources/META-INF/spring.tooling

@@ -0,0 +1,4 @@
+# Tooling related information for the integration ip namespace
+http\://www.springframework.org/schema/integration/ip@name=integration ip Namespace
+http\://www.springframework.org/schema/integration/ip@prefix=int-ip
+http\://www.springframework.org/schema/integration/ip@icon=org/springframework/integration/ip/config/spring-integration-ip.gif

org.springframework.integration.ip/src/main/resources/log4j.properties

@@ -0,0 +1,8 @@
+log4j.rootCategory=DEBUG, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n
+
+log4j.category.org.springframework.integration=DEBUG
+log4j.category.org.springframework.integration.file=DEBUG