From 8bf8caa22ea09d60c257538dcdb543281108317d Mon Sep 17 00:00:00 2001 From: Marcin Pilaczynski Date: Mon, 9 Nov 2015 22:09:02 +0100 Subject: [PATCH] INT-3883: UDP: destination and socket expressions JIRA: https://jira.spring.io/browse/INT-3883 UDP Outbound Channel Adapter is able to use given UDP Inbound Adapters server socket (outgoing packets will have source port same as incoming packets destination port). * Introduce `destinationExpression` instead of hard-coded `host/port` logic in the `DatagramPacketMessageMapper` * Make `UnicastSendingMessageHandler.destinationExpression` mutually exclusive with `host/port` pair * Move `socketExpression` to the setter as it is absolutely different option from the `destination` * Make `UnicastSendingMessageHandler` expressions logic based on the `requestMessage` INT-3883 Code review fixes. Further polishing Polishing according PR comments Rework the String socket address logic just to the expected `URI` style. Accept `2016` for changed classes. --- .../integration/ip/IpHeaders.java | 8 +- .../ip/config/IpAdapterParserUtils.java | 54 ++++--- .../UdpInboundChannelAdapterParser.java | 14 +- .../UdpOutboundChannelAdapterParser.java | 36 +++-- .../ip/udp/DatagramPacketMessageMapper.java | 8 +- .../udp/MulticastReceivingChannelAdapter.java | 7 +- .../udp/MulticastSendingMessageHandler.java | 8 +- .../udp/UnicastReceivingChannelAdapter.java | 4 +- .../ip/udp/UnicastSendingMessageHandler.java | 132 +++++++++++++++--- .../ip/config/spring-integration-ip-4.3.xsd | 23 ++- ...ramPacketMulticastSendingHandlerTests.java | 7 +- .../DatagramPacketSendingHandlerTests.java | 3 +- .../ip/udp/UdpChannelAdapterTests.java | 25 +++- .../udp/testIp-socket-expression-context.xml | 22 +++ src/reference/asciidoc/ip.adoc | 46 +++++- src/reference/asciidoc/whats-new.adoc | 4 + 16 files changed, 318 insertions(+), 83 deletions(-) create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-socket-expression-context.xml diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java index c852d4d36b..86d538aec4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2016 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. @@ -22,6 +22,7 @@ package org.springframework.integration.ip; * @author Mark Fisher * @author Gary Russell * @author Dave Syer + * @author Artem Bilan * @since 2.0 */ public final class IpHeaders { @@ -46,6 +47,11 @@ public final class IpHeaders { */ public static final String PORT = IP + "port"; + /** + * The remote address for a UDP packet. + */ + public static final String PACKET_ADDRESS = IP + "packetAddress"; + /** * The remote ip address to which UDP application-level acks will be sent. The * framework includes acknowledgment information in the data packet. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 8f06249a43..77fa2ed808 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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. @@ -28,6 +28,8 @@ import org.springframework.util.StringUtils; * Utility methods and constants for IP adapter parsers. * * @author Gary Russell + * @author Marcin Pilaczynski + * @author Artem Bilan * @since 2.0 */ public abstract class IpAdapterParserUtils { @@ -70,8 +72,6 @@ public abstract class IpAdapterParserUtils { static final String USING_DIRECT_BUFFERS = "using-direct-buffers"; - static final String MESSAGE_FORMAT = "message-format"; - static final String SO_LINGER = "so-linger"; static final String SO_TCP_NODELAY = "so-tcp-no-delay"; @@ -136,9 +136,16 @@ public abstract class IpAdapterParserUtils { * @param attributeName the name of the attribute whose value will be * @param trueFalse not used * used to populate the property + * @deprecated in favor of {@link #addConstructorValueIfAttributeDefined}. */ + @Deprecated public static void addConstuctorValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName, boolean trueFalse) { + addConstructorValueIfAttributeDefined(builder, element, attributeName); + } + + public static void addConstructorValueIfAttributeDefined(BeanDefinitionBuilder builder, + Element element, String attributeName) { String attributeValue = element.getAttribute(attributeName); if (StringUtils.hasText(attributeValue)) { builder.addConstructorArgValue(attributeValue); @@ -146,11 +153,35 @@ public abstract class IpAdapterParserUtils { } /** + * Adds destination configuration to constructor. + * * @param element The element. * @param builder The builder. * @param parserContext The parser context. */ - public static void addHostAndPortToConstructor(Element element, + public static void addDestinationConfigToConstructor(Element element, + BeanDefinitionBuilder builder, ParserContext parserContext) { + String destinationExpression = element.getAttribute("destination-expression"); + if (StringUtils.hasText(destinationExpression)) { + addDestinationExpressionToConstructor(destinationExpression, element, builder, parserContext); + } + else { + addHostAndPortToConstructor(element, builder, parserContext); + } + } + + private static void addDestinationExpressionToConstructor(String destinationExpression, + Element element, BeanDefinitionBuilder builder, ParserContext parserContext) { + String host = element.getAttribute(IpAdapterParserUtils.HOST); + String port = element.getAttribute(IpAdapterParserUtils.PORT); + if (StringUtils.hasText(host) || StringUtils.hasText(port)) { + parserContext.getReaderContext() + .error("The 'destination-expression' is mutually exclusive with 'host/port' pair.", element); + } + builder.addConstructorArgValue(destinationExpression); + } + + private static void addHostAndPortToConstructor(Element element, BeanDefinitionBuilder builder, ParserContext parserContext) { String host = element.getAttribute(IpAdapterParserUtils.HOST); if (!StringUtils.hasText(host)) { @@ -162,17 +193,6 @@ public abstract class IpAdapterParserUtils { builder.addConstructorArgValue(port); } - /** - * @param element The element. - * @param builder The builder. - * @param parserContext The parser context. - */ - public static void addPortToConstructor(Element element, - BeanDefinitionBuilder builder, ParserContext parserContext) { - String port = IpAdapterParserUtils.getPort(element, parserContext); - builder.addConstructorArgValue(port); - } - /** * Asserts that a port attribute is supplied. * @param element The element. @@ -183,8 +203,8 @@ public abstract class IpAdapterParserUtils { static String getPort(Element element, ParserContext parserContext) { String port = element.getAttribute(IpAdapterParserUtils.PORT); if (!StringUtils.hasText(port)) { - parserContext.getReaderContext().error(IpAdapterParserUtils.PORT + - " is required for IP channel adapters", element); + parserContext.getReaderContext().error(IpAdapterParserUtils.PORT + + " is required for IP channel adapters", element); } return port; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java index bab65eca2c..0a852b7b24 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2016 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. @@ -16,6 +16,8 @@ package org.springframework.integration.ip.config; +import org.w3c.dom.Element; + import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; @@ -24,12 +26,12 @@ 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; -import org.w3c.dom.Element; /** * Channel Adapter that receives UDP datagram packets and maps them to Messages. * * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ public class UdpInboundChannelAdapterParser extends AbstractChannelAdapterParser { @@ -52,11 +54,6 @@ public class UdpInboundChannelAdapterParser extends AbstractChannelAdapterParser return builder.getBeanDefinition(); } - /** - * @param element - * @param builder - * @param parserContext - */ private void addPortToConstructor(Element element, BeanDefinitionBuilder builder, ParserContext parserContext) { String port = IpAdapterParserUtils.getPort(element, parserContext); @@ -82,8 +79,7 @@ public class UdpInboundChannelAdapterParser extends AbstractChannelAdapterParser builder.addConstructorArgValue(mcAddress); } addPortToConstructor(element, builder, parserContext); - IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.CHECK_LENGTH, true); + IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH); return builder; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpOutboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpOutboundChannelAdapterParser.java index 0cb1cdac80..18ae4b4d29 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpOutboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2016 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. @@ -16,22 +16,25 @@ package org.springframework.integration.ip.config; +import org.w3c.dom.Element; + 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; -import org.w3c.dom.Element; /** * @author Gary Russell + * @author Marcin Pilaczynski + * @author Artem Bilan * @since 2.0 */ public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { - private static final String BASE_PACKAGE = "org.springframework.integration.ip.udp"; - protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = parseUdp(element, parserContext); IpAdapterParserUtils.addCommonSocketOptions(builder, element); @@ -42,8 +45,7 @@ public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap BeanDefinitionBuilder builder; String multicast = IpAdapterParserUtils.getMulticast(element); if (multicast.equals("true")) { - builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + - ".MulticastSendingMessageHandler"); + builder = BeanDefinitionBuilder.genericBeanDefinition(MulticastSendingMessageHandler.class); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.MIN_ACKS_SUCCESS, "minAcksForSuccess"); @@ -52,20 +54,14 @@ public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap "timeToLive"); } else { - builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + - ".UnicastSendingMessageHandler"); + builder = BeanDefinitionBuilder.genericBeanDefinition(UnicastSendingMessageHandler.class); } - IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext); - IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.CHECK_LENGTH, true); - IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.ACK, true); - IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.ACK_HOST, false); - IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.ACK_PORT, false); - IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.ACK_TIMEOUT, false); + IpAdapterParserUtils.addDestinationConfigToConstructor(element, builder, parserContext); + IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH); + IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK); + IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_HOST); + IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_PORT); + IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_TIMEOUT); String ack = element.getAttribute(IpAdapterParserUtils.ACK); if (ack.equals("true")) { if (!StringUtils.hasText(element @@ -86,6 +82,8 @@ public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap IpAdapterParserUtils.RECEIVE_BUFFER_SIZE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, IpAdapterParserUtils.TASK_EXECUTOR); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + "socket-expression", "socketExpressionString"); return builder; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java index 9d5bc610ca..f66f017e27 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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. @@ -62,8 +62,8 @@ import org.springframework.util.Assert; * @author Artem Bilan * @since 2.0 */ -public class DatagramPacketMessageMapper implements InboundMessageMapper, OutboundMessageMapper, - BeanFactoryAware { +public class DatagramPacketMessageMapper implements InboundMessageMapper, + OutboundMessageMapper, BeanFactoryAware { private volatile String charset = "UTF-8"; @@ -234,6 +234,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper message) throws Exception { + super.convertAndSend(message); if (logger.isDebugEnabled()) { logger.debug("Sent packet to " + this.multicastSocket.getInterface()); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 3fb1eff4b7..8c3f989af7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -218,7 +218,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece return this.socket; } - protected synchronized DatagramSocket getSocket() { + public synchronized DatagramSocket getSocket() { if (this.socket == null) { try { DatagramSocket socket = null; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index e8d60c7c27..4f79ba3f74 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2015 the original author or authors. + * Copyright 2001-2016 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. @@ -21,7 +21,9 @@ import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.SocketException; +import java.net.URI; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -33,7 +35,9 @@ import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.springframework.integration.MessageRejectedException; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; @@ -50,6 +54,8 @@ import org.springframework.util.Assert; * a UDP acknowledgment to confirm delivery. * * @author Gary Russell + * @author Marcin Pilaczynski + * @author Artem Bilan * @since 2.0 */ public class UnicastSendingMessageHandler extends @@ -57,8 +63,9 @@ public class UnicastSendingMessageHandler extends private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); - private volatile DatagramSocket socket; + private final Expression destinationExpression; + private volatile DatagramSocket socket; /** * If true adds headers to instruct receiving adapter to return an ack. @@ -90,6 +97,10 @@ public class UnicastSendingMessageHandler extends private volatile boolean taskExecutorSet; + private Expression socketExpression; + + private EvaluationContext evaluationContext; + /** * Basic constructor; no reliability; no acknowledgment. * @param host Destination host. @@ -99,6 +110,37 @@ public class UnicastSendingMessageHandler extends super(host, port); this.mapper.setLengthCheck(false); this.mapper.setAcknowledge(false); + this.destinationExpression = null; + } + + /** + * Construct UnicastSendingMessageHandler based on the destination SpEL expression to determine + * the target destination at runtime against requestMessage. + * @param destinationExpression the SpEL expression to evaluate the target destination at runtime. + * Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. + * @since 4.3 + */ + public UnicastSendingMessageHandler(String destinationExpression) { + super("", 0); + Assert.hasText(destinationExpression, "'destinationExpression' cannot be null or empty"); + this.mapper.setLengthCheck(false); + this.mapper.setAcknowledge(false); + this.destinationExpression = EXPRESSION_PARSER.parseExpression(destinationExpression); + } + + /** + * Construct UnicastSendingMessageHandler based on the destination SpEL expression to determine + * the target destination at runtime against requestMessage. + * @param destinationExpression the SpEL expression to evaluate the target destination at runtime. + * Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. + * @since 4.3 + */ + public UnicastSendingMessageHandler(Expression destinationExpression) { + super("", 0); + Assert.notNull(destinationExpression, "'destinationExpression' cannot be null"); + this.mapper.setLengthCheck(false); + this.mapper.setAcknowledge(false); + this.destinationExpression = destinationExpression; } /** @@ -111,6 +153,7 @@ public class UnicastSendingMessageHandler extends super(host, port); this.mapper.setLengthCheck(lengthCheck); this.mapper.setAcknowledge(false); + this.destinationExpression = null; } /** @@ -129,6 +172,7 @@ public class UnicastSendingMessageHandler extends int ackPort, int ackTimeout) { super(host, port); + this.destinationExpression = null; setReliabilityAttributes(false, acknowledge, ackHost, ackPort, ackTimeout); } @@ -151,6 +195,7 @@ public class UnicastSendingMessageHandler extends int ackPort, int ackTimeout) { super(host, port); + this.destinationExpression = null; setReliabilityAttributes(lengthCheck, acknowledge, ackHost, ackPort, ackTimeout); } @@ -179,6 +224,7 @@ public class UnicastSendingMessageHandler extends Executor executor = Executors .newSingleThreadExecutor(new ThreadFactory() { private final AtomicInteger n = new AtomicInteger(); + @Override public Thread newThread(Runnable runner) { Thread thread = new Thread(runner); @@ -186,6 +232,7 @@ public class UnicastSendingMessageHandler extends thread.setDaemon(true); return thread; } + }); this.taskExecutor = executor; } @@ -203,30 +250,26 @@ public class UnicastSendingMessageHandler extends } @Override - public void handleMessageInternal(Message message) - throws MessageRejectedException, MessageHandlingException, + public void handleMessageInternal(Message message) throws MessageHandlingException, MessageDeliveryException { if (this.acknowledge) { - Assert.state(this.isRunning(), "When 'acknowlege' is enabled, adapter must be running"); + Assert.state(this.isRunning(), "When 'acknowledge' is enabled, adapter must be running"); startAckThread(); } CountDownLatch countdownLatch = null; String messageId = message.getHeaders().getId().toString(); try { - DatagramPacket packet; - if (this.waitForAck) { + boolean waitForAck = this.waitForAck; + if (waitForAck) { countdownLatch = new CountDownLatch(ackCounter); this.ackControl.put(messageId, countdownLatch); } - packet = this.mapper.fromMessage(message); - this.send(packet); - if (logger.isDebugEnabled()) { - logger.debug("Sent packet for message " + message + " to " + packet.getSocketAddress()); - } - if (this.waitForAck) { + convertAndSend(message); + if (waitForAck) { try { if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) { - throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis"); + throw new MessagingException(message, "Failed to receive UDP Ack in " + + this.ackTimeout + " millis"); } } catch (InterruptedException e) { @@ -275,10 +318,41 @@ public class UnicastSendingMessageHandler extends } } - protected void send(DatagramPacket packet) throws Exception { - DatagramSocket socket = this.getSocket(); - packet.setSocketAddress(this.getDestinationAddress()); + protected void convertAndSend(Message message) throws Exception { + DatagramSocket socket; + if (this.socketExpression != null) { + socket = this.socketExpression.getValue(this.evaluationContext, message, DatagramSocket.class); + } + else { + socket = getSocket(); + } + SocketAddress destinationAddress; + if (this.destinationExpression != null) { + Object destination = this.destinationExpression.getValue(this.evaluationContext, message); + if (destination instanceof String) { + destination = new URI((String) destination); + } + if (destination instanceof URI) { + URI uri = (URI) destination; + destination = new InetSocketAddress(uri.getHost(), uri.getPort()); + } + if (destination instanceof SocketAddress) { + destinationAddress = (SocketAddress) destination; + } + else { + throw new IllegalStateException("'destinationExpression' must evaluate to String, URI " + + "or SocketAddress. Gotten [" + destination + "]."); + } + } + else { + destinationAddress = getDestinationAddress(); + } + DatagramPacket packet = this.mapper.fromMessage(message); + packet.setSocketAddress(destinationAddress); socket.send(packet); + if (logger.isDebugEnabled()) { + logger.debug("Sent packet for message " + message + " to " + packet.getSocketAddress()); + } } protected void setSocket(DatagramSocket socket) { @@ -346,6 +420,22 @@ public class UnicastSendingMessageHandler extends this.ackCounter = ackCounter; } + /** + * @param socketExpression the socket expression to determine the target socket at runtime. + * @since 4.3 + */ + public void setSocketExpression(Expression socketExpression) { + this.socketExpression = socketExpression; + } + + /** + * @param socketExpression the socket SpEL expression to determine the target socket at runtime. + * @since 4.3 + */ + public void setSocketExpressionString(String socketExpression) { + this.socketExpression = EXPRESSION_PARSER.parseExpression(socketExpression); + } + @Override public String getComponentType(){ return "ip:udp-outbound-channel-adapter"; @@ -381,7 +471,11 @@ public class UnicastSendingMessageHandler extends @Override protected void onInit() throws Exception { super.onInit(); - this.mapper.setBeanFactory(this.getBeanFactory()); + this.mapper.setBeanFactory(getBeanFactory()); + this.evaluationContext = IntegrationContextUtils.getEvaluationContext(getBeanFactory()); + if (this.socketExpression != null) { + Assert.state(!this.acknowledge, "'acknowledge' must be false when using a socket expression"); + } } protected void setSocketAttributes(DatagramSocket socket) throws SocketException { diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.3.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.3.xsd index ab674ca456..176b5004bd 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.3.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.3.xsd @@ -80,8 +80,29 @@ - + + + + + A SpEL expression to determine the destination address for the datagram packet + at runtime. The 'requestMessage' is used as a root object for evaluation context. + Must evaluate to 'URI', or 'String' in the URI style or 'SocketAddress'. + The 'IpHeaders.PACKET_ADDRESS' header from the received message throughout + 'udp-inbound-channel-adapter' can be used with this expression as well. + Mutually exclusive with 'host'/'port' pair. + + + + + + + A SpEL expression to be evaluated to determine which datagram + socket use for sending outgoing UDP packets (e.g. UDP inbound + packet receiving Channel Adapter socket can be used). + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java index a9a49f188d..d8f4469728 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -18,6 +18,7 @@ package org.springframework.integration.ip.udp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; import java.net.DatagramPacket; import java.net.DatagramSocket; @@ -34,6 +35,7 @@ import org.apache.commons.logging.LogFactory; import org.junit.Rule; import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -98,7 +100,9 @@ public class DatagramPacketMulticastSendingHandlerTests { executor.execute(catcher); assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort); + handler.setBeanFactory(mock(BeanFactory.class)); handler.setLocalAddress(this.multicastRule.getNic()); + handler.afterPropertiesSet(); handler.handleMessage(MessageBuilder.withPayload(payload).build()); assertTrue(received.await(10000, TimeUnit.MILLISECONDS)); handler.stop(); @@ -176,6 +180,7 @@ public class DatagramPacketMulticastSendingHandlerTests { new MulticastSendingMessageHandler(multicastAddress, testPort, true, true, "localhost", 0, 10000); handler.setLocalAddress(this.multicastRule.getNic()); handler.setMinAcksForSuccess(2); + handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); handler.start(); waitAckListening(handler); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java index 5d76a83cc2..8de31f6aaf 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -38,6 +38,7 @@ import org.springframework.messaging.Message; /** * @author Mark Fisher * @author Gary Russell + * @author Marcin Pilaczynski * @since 2.0 */ public class DatagramPacketSendingHandlerTests { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index 52eabe772f..b3c6066509 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -39,6 +39,8 @@ import org.junit.Rule; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.ServiceActivatingHandler; @@ -53,6 +55,7 @@ import org.springframework.messaging.SubscribableChannel; * * @author Gary Russell * @author Artem Bilan + * @author Marcin Pilaczynski * @since 2.0 * */ @@ -326,12 +329,30 @@ public class UdpChannelAdapterTests { adapter.stop(); } + @Test + public void testSocketExpression() throws Exception { + ConfigurableApplicationContext context = + new ClassPathXmlApplicationContext("testIp-socket-expression-context.xml", getClass()); + UnicastReceivingChannelAdapter adapter = context.getBean(UnicastReceivingChannelAdapter.class); + SocketTestUtils.waitListening(adapter); + int receiverServerPort = adapter.getPort(); + DatagramPacket packet = new DatagramPacket("foo".getBytes(), 3); + packet.setSocketAddress(new InetSocketAddress("localhost", receiverServerPort)); + DatagramSocket socket = new DatagramSocket(); + socket.send(packet); + socket.receive(packet); + assertEquals("FOO", new String(packet.getData())); + assertEquals(receiverServerPort, packet.getPort()); + context.close(); + } + private class FailingService { + @SuppressWarnings("unused") public String serviceMethod(byte[] bytes) { throw new RuntimeException("Failed"); } + } - } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-socket-expression-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-socket-expression-context.xml new file mode 100644 index 0000000000..d3dd36c229 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-socket-expression-context.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 5b3431ed72..9d0c134f99 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -99,6 +99,43 @@ For even more reliable networking, TCP can be used. Starting with _version 4.3_, the `ackPort` can be set to `0`, in which case the Operating System chooses the port. +Also starting with _version 4.3_, the `destination-expression` and `socket-expression` options are available +for the `` (`UnicastSendingMessageHandler`). + +The `destination-expression` can be used as a runtime alternative to the hardcoded `host`/`port` pair to determine +the destination address for the outgoing datagram packet against `requestMessage` as a root object for evaluation context. +The expression must evaluate to `URI`, or `String` in the URI style (see http://www.ietf.org/rfc/rfc2396.txt[RFC-2396]) +or `SocketAddress`. +The new `IpHeaders.PACKET_ADDRESS` header can be used for this expression as well. +In the Framework this header is populated by the `DatagramPacketMessageMapper`, when we receive datagrams in the +`UnicastReceivingChannelAdapter` and convert them to messages. +The header value is exactly the result of `DatagramPacket.getSocketAddress()` of incoming datagram. + +With the `socket-expression` help the Outbound Channel Adapter can use e.g. Inbound Channel Adapter socket +to send datagrams through same port which they were received. +It's useful in a scenario when our application works as a UDP server and clients operate behind the NAT. +This expression must evaluate to the `DatagramSocket`. +The `requestMessage` is used as a root object for evaluation context. +The `socket-expression` parameter cannot be used with parameters like `multicast` and `acknowledge`. + +[source,xml] +---- + + + + + + + + + +---- + + [source,xml] ---- > for more information. + +The `destination-expression` and `socket-expression` are now available for the ``. +See <> for more information. +