INT-782, INT-783 initial (rough) commit of UDP sending and receiving adapters

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2002-2009 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.udp;
+
+import java.io.UnsupportedEncodingException;
+import java.net.DatagramPacket;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.message.InboundMessageMapper;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.OutboundMessageMapper;
+
+/**
+ * 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.
+ * 
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>,
+		OutboundMessageMapper<DatagramPacket> {
+
+	private volatile String charset = "UTF-8";
+
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public DatagramPacket fromMessage(Message<?> message) throws Exception {
+		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 new DatagramPacket(bytes, bytes.length);
+	}
+
+	public Message<byte[]> toMessage(DatagramPacket packet) throws Exception {
+		int offset = packet.getOffset();
+		int length = packet.getLength();
+		byte[] payload = new byte[length];
+		System.arraycopy(packet.getData(), offset, payload, 0, length);
+		Message<byte[]> message = null;
+		if (payload.length > 0) {
+			message = MessageBuilder.withPayload(payload)
+					.setHeader(UdpHeaders.HOSTNAME, packet.getAddress().getHostName())
+					.setHeader(UdpHeaders.IP_ADDRESS, packet.getAddress().getHostAddress())
+					.build();
+		}
+		return message;
+	}
+
+}

org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapter.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2002-2009 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.udp;
+
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.util.Date;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.Assert;
+
+/**
+ * Channel Adapter that receives UDP datagram packets and maps them to Messages.
+ * 
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class DatagramPacketReceivingChannelAdapter extends MessageProducerSupport implements Runnable {
+
+	private final int port;
+
+	private volatile int timeout = 60 * 1000;
+
+	private volatile int receiveBufferSize = 64 * 1024;
+
+	private volatile boolean active;
+
+	private volatile DatagramSocket socket;
+
+	private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
+
+
+	public DatagramPacketReceivingChannelAdapter(int port) {
+		this.port = port;
+	}
+
+
+	public void setReceiveBufferSize(int receiveBufferSize) {
+		this.receiveBufferSize = receiveBufferSize;
+	}
+
+	public void setTimeout(int timeout) {
+		this.timeout = timeout;
+		if (this.socket != null) {
+			try {
+				this.socket.setSoTimeout(timeout);
+			}
+			catch (SocketException e) {
+				throw new IllegalStateException("failed to set socket timeout", e);
+			}
+		}
+	}
+
+	@Override
+	protected void doStart() {
+		TaskScheduler taskScheduler = this.getTaskScheduler();
+		Assert.state(taskScheduler != null, "taskScheduler is required");
+		this.active = true;
+		taskScheduler.schedule(this, new Date());
+	}
+
+	@Override
+	protected void doStop() {
+		this.active = false;
+		try {
+			this.socket.close();
+		}
+		catch (Exception e) {
+			// ignore
+		}
+	}
+
+	public void run() {
+		while (this.active) {
+			try {
+				Message<byte[]> message = receive();
+				if (message != null) {
+					this.sendMessage(message);
+				}
+			}
+			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);
+			}
+		}
+	}
+
+	public Message<byte[]> 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 this.mapper.toMessage(packet);
+	}
+
+	private synchronized DatagramSocket getSocket() {
+		if (this.socket == null) {
+			try {
+				this.socket = new DatagramSocket(this.port);
+				this.socket.setSoTimeout(this.timeout);
+			}
+			catch (SocketException e) {
+				throw new MessagingException("failed to create DatagramSocket", e);
+			}
+		}
+		return this.socket;
+	}
+
+}

org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketSendingHandler.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2002-2009 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.udp;
+
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.OutboundMessageMapper;
+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.
+ * 
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class DatagramPacketSendingHandler implements MessageHandler {
+
+	private final SocketAddress socketAddress;
+
+	private final OutboundMessageMapper<DatagramPacket> mapper = new DatagramPacketMessageMapper();
+
+
+	public DatagramPacketSendingHandler(String host, int port) {
+		Assert.notNull(host, "host must not be null");
+		this.socketAddress = new InetSocketAddress(host, port);
+	}
+
+
+	public void handleMessage(Message<?> message) {
+		try {
+			DatagramPacket packet = this.mapper.fromMessage(message);
+			this.send(packet);
+		}
+		catch (MessagingException e) {
+			throw e;
+		}
+		catch (Exception e) {
+			throw new MessageHandlingException(message, "failed to send UDP packet", e);
+		}
+	}
+
+	private void send(DatagramPacket packet) throws Exception {
+		DatagramSocket socket = null;
+		try {
+			socket = new DatagramSocket();
+			packet.setSocketAddress(this.socketAddress);
+			socket.send(packet);
+		}
+		finally {
+			if (socket != null) {
+				socket.close();
+			}
+		}
+	}
+
+}

org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/UdpHeaders.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2002-2009 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.udp;
+
+/**
+ * Headers for Messages mapped from UDP datagram packets.
+ * 
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public abstract class UdpHeaders {
+
+	private static final String PREFIX = "springintegration_udp_";
+
+	public static final String HOSTNAME = PREFIX + "hostname";
+
+	public static final String IP_ADDRESS = PREFIX + "ip"; 
+
+}

org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapterTests.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2002-2009 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.udp;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetSocketAddress;
+
+import org.junit.Test;
+
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class DatagramPacketReceivingChannelAdapterTests {
+
+	@Test
+	public void receive() throws IOException {
+		int testPort = 23487;
+		QueueChannel output = new QueueChannel();
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		DatagramPacketReceivingChannelAdapter adapter = new DatagramPacketReceivingChannelAdapter(testPort);
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.setOutputChannel(output);
+		adapter.afterPropertiesSet();
+		DatagramSocket socket = new DatagramSocket();
+		byte[] bytes = "foo".getBytes("UTF-8");
+		DatagramPacket packet = new DatagramPacket(bytes, bytes.length);
+		packet.setSocketAddress(new InetSocketAddress(testPort));
+		socket.send(packet);
+		socket.close();
+		Message<?> message = output.receive(3000);
+		assertNotNull(message);
+		assertEquals("foo", new String((byte[]) message.getPayload(), "UTF-8"));
+	}
+
+}

org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketSendingHandlerTests.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2002-2009 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.udp;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import org.springframework.integration.message.MessageBuilder;
+
+/**
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class DatagramPacketSendingHandlerTests {
+
+	@Test
+	public void verifySend() throws Exception {
+		final int testPort = 27816;
+		byte[] buffer = new byte[8];
+		final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
+		final CountDownLatch latch = new CountDownLatch(1);
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			public void run() {
+				try {
+					DatagramSocket socket = new DatagramSocket(testPort);
+					socket.receive(receivedPacket);
+					latch.countDown();
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		DatagramPacketSendingHandler handler = new DatagramPacketSendingHandler("localhost", testPort);
+		String payload = "foo";
+		handler.handleMessage(MessageBuilder.withPayload(payload).build());
+		latch.await(3000, TimeUnit.MILLISECONDS);
+		byte[] src = receivedPacket.getData();
+		int length = receivedPacket.getLength();
+		int offset = receivedPacket.getOffset();
+		byte[] dest = new byte[length];
+		System.arraycopy(src, offset, dest, 0, length);
+		assertEquals(payload, new String(dest));
+	}
+
+}

org.springframework.integration.udp/template.mf

@@ -8,5 +8,6 @@ Import-Template:
  org.springframework.beans.*;version="[3.0.0, 4.0.0)",
  org.springframework.context;version="[3.0.0, 4.0.0)",
  org.springframework.core.*;version="[3.0.0, 4.0.0)",
+ org.springframework.scheduling.*;version="[3.0.0, 4.0.0)",
  org.springframework.util;version="[3.0.0, 4.0.0)",
  org.w3c.dom.*;version="0"