diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java new file mode 100644 index 0000000000..b5e2e8b15b --- /dev/null +++ b/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()); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java new file mode 100644 index 0000000000..a8e3b12fcf --- /dev/null +++ b/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; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java new file mode 100644 index 0000000000..28f7f230c6 --- /dev/null +++ b/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); + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java new file mode 100644 index 0000000000..e21143037e --- /dev/null +++ b/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"; + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java new file mode 100644 index 0000000000..e0991ee47c --- /dev/null +++ b/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); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java new file mode 100644 index 0000000000..0f3f1316e1 --- /dev/null +++ b/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(); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java new file mode 100644 index 0000000000..2647a39e00 --- /dev/null +++ b/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 ip 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()); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..7a4e420d42 --- /dev/null +++ b/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(); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java new file mode 100644 index 0000000000..57479fb9c5 --- /dev/null +++ b/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: + * + * + * @author Mark Fisher + * @author Gary Russell + * @since 2.0 + */ +public class DatagramPacketMessageMapper implements InboundMessageMapper, + OutboundMessageMapper { + + 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 toMessage(DatagramPacket packet) throws Exception { + int offset = packet.getOffset(); + int length = packet.getLength(); + byte[] payload; + ByteBuffer buffer = ByteBuffer.wrap(packet.getData(), offset, length); + Message 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); + } + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java new file mode 100644 index 0000000000..44d13d522a --- /dev/null +++ b/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; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java new file mode 100644 index 0000000000..bac3537b70 --- /dev/null +++ b/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; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UdpSocket.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UdpSocket.java new file mode 100644 index 0000000000..fe38c210d5 --- /dev/null +++ b/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(); + } + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java new file mode 100644 index 0000000000..a4251bd466 --- /dev/null +++ b/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 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 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 + } + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java new file mode 100644 index 0000000000..59bb1c8891 --- /dev/null +++ b/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 ackControl = Collections + .synchronizedMap(new HashMap()); + + 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(); + } + } + +} diff --git a/org.springframework.integration.ip/src/main/resources/META-INF/spring.handlers b/org.springframework.integration.ip/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..8071ccd16c --- /dev/null +++ b/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 \ No newline at end of file diff --git a/org.springframework.integration.ip/src/main/resources/META-INF/spring.schemas b/org.springframework.integration.ip/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..5cf91575f4 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/org.springframework.integration.ip/src/main/resources/META-INF/spring.tooling b/org.springframework.integration.ip/src/main/resources/META-INF/spring.tooling new file mode 100644 index 0000000000..caed74cd0c --- /dev/null +++ b/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 diff --git a/org.springframework.integration.ip/src/main/resources/log4j.properties b/org.springframework.integration.ip/src/main/resources/log4j.properties new file mode 100644 index 0000000000..d35f84941d --- /dev/null +++ b/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 \ No newline at end of file