From 39069b331f66f43af53be7017ab2c5fd27e0164b Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 9 Oct 2009 16:56:15 +0000 Subject: [PATCH] INT-782, INT-783 initial (rough) commit of UDP sending and receiving adapters --- .../udp/DatagramPacketMessageMapper.java | 85 +++++++++++ ...DatagramPacketReceivingChannelAdapter.java | 136 ++++++++++++++++++ .../udp/DatagramPacketSendingHandler.java | 78 ++++++++++ .../integration/udp/UdpHeaders.java | 33 +++++ ...ramPacketReceivingChannelAdapterTests.java | 60 ++++++++ .../DatagramPacketSendingHandlerTests.java | 67 +++++++++ .../template.mf | 1 + 7 files changed, 460 insertions(+) create mode 100644 org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketMessageMapper.java create mode 100644 org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapter.java create mode 100644 org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketSendingHandler.java create mode 100644 org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/UdpHeaders.java create mode 100644 org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapterTests.java create mode 100644 org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketSendingHandlerTests.java diff --git a/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketMessageMapper.java b/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketMessageMapper.java new file mode 100644 index 0000000000..a3286fe547 --- /dev/null +++ b/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, + OutboundMessageMapper { + + 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 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 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; + } + +} diff --git a/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapter.java b/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapter.java new file mode 100644 index 0000000000..9559b13508 --- /dev/null +++ b/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 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 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; + } + +} diff --git a/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketSendingHandler.java b/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/DatagramPacketSendingHandler.java new file mode 100644 index 0000000000..d60b8a67fd --- /dev/null +++ b/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 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(); + } + } + } + +} diff --git a/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/UdpHeaders.java b/org.springframework.integration.udp/src/main/java/org/springframework/integration/udp/UdpHeaders.java new file mode 100644 index 0000000000..139ee84c40 --- /dev/null +++ b/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"; + +} diff --git a/org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapterTests.java b/org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketReceivingChannelAdapterTests.java new file mode 100644 index 0000000000..115b74744c --- /dev/null +++ b/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")); + } + +} diff --git a/org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketSendingHandlerTests.java b/org.springframework.integration.udp/src/test/java/org/springframework/integration/udp/DatagramPacketSendingHandlerTests.java new file mode 100644 index 0000000000..cac0683a36 --- /dev/null +++ b/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)); + } + +} diff --git a/org.springframework.integration.udp/template.mf b/org.springframework.integration.udp/template.mf index 3188006dd1..97b4eff8de 100644 --- a/org.springframework.integration.udp/template.mf +++ b/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"