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.

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.

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 +
+			parserContext.getReaderContext().error(IpAdapterParserUtils.PORT
-					" is required for IP channel adapters", element);
+					+ " is required for IP channel adapters", element);
 		}
 		return port;
 	}

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,
+		IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH);
-				element, IpAdapterParserUtils.CHECK_LENGTH, true);
 		return builder;
 	}
 

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 +
+			builder = BeanDefinitionBuilder.genericBeanDefinition(MulticastSendingMessageHandler.class);
-					".MulticastSendingMessageHandler");
 			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 +
+			builder = BeanDefinitionBuilder.genericBeanDefinition(UnicastSendingMessageHandler.class);
-					".UnicastSendingMessageHandler");
 		}
-		IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
+		IpAdapterParserUtils.addDestinationConfigToConstructor(element, builder, parserContext);
-		IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
+		IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH);
-				element, IpAdapterParserUtils.CHECK_LENGTH, true);
+		IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK);
-		IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
+		IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_HOST);
-				element, IpAdapterParserUtils.ACK, true);
+		IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_PORT);
-		IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
+		IpAdapterParserUtils.addConstructorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_TIMEOUT);
-				element, IpAdapterParserUtils.ACK_HOST, false);
-		IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
-				element, IpAdapterParserUtils.ACK_PORT, false);
-		IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
-				element, IpAdapterParserUtils.ACK_TIMEOUT, false);
 		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;
 	}
 

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<DatagramPacket>, OutboundMessageMapper<DatagramPacket>,
+public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>,
-		BeanFactoryAware {
+		OutboundMessageMapper<DatagramPacket>, BeanFactoryAware {
 
 	private volatile String charset = "UTF-8";
 
@@ -234,6 +234,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
 							.setHeader(IpHeaders.HOSTNAME, hostName)
 							.setHeader(IpHeaders.IP_ADDRESS, hostAddress)
 							.setHeader(IpHeaders.PORT, port)
+							.setHeader(IpHeaders.PACKET_ADDRESS, packet.getSocketAddress())
 							.build();
 				}  // on no match, just treat as simple payload
 			}
@@ -249,6 +250,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
 						.setHeader(IpHeaders.HOSTNAME, hostName)
 						.setHeader(IpHeaders.IP_ADDRESS, hostAddress)
 						.setHeader(IpHeaders.PORT, port)
+						.setHeader(IpHeaders.PACKET_ADDRESS, packet.getSocketAddress())
 						.build();
 			}
 		}

spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.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.
@@ -28,6 +28,7 @@ import org.springframework.messaging.MessagingException;
  * sends them to an output channel.
  *
  * @author Gary Russell
+ * @author Marcin Pilaczynski
  * @since 2.0
  */
 public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAdapter {
@@ -60,8 +61,8 @@ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAda
 	}
 
 	@Override
-	protected synchronized DatagramSocket getSocket() {
+	public synchronized DatagramSocket getSocket() {
-		if (this.getTheSocket() == null) {
+		if (getTheSocket() == null) {
 			try {
 				int port = getPort();
 				MulticastSocket socket = port == 0 ? new MulticastSocket() : new MulticastSocket(port);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2014 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.
@@ -17,12 +17,12 @@
 package org.springframework.integration.ip.udp;
 
 import java.io.IOException;
-import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.MulticastSocket;
 
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 
 /**
@@ -171,8 +171,8 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
 	}
 
 	@Override
-	protected void send(DatagramPacket packet) throws Exception {
+	protected void convertAndSend(Message<?> message) throws Exception {
-		super.send(packet);
+		super.convertAndSend(message);
 		if (logger.isDebugEnabled()) {
 			logger.debug("Sent packet to " + this.multicastSocket.getInterface());
 		}

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;

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)
+	public void handleMessageInternal(Message<?> message) throws MessageHandlingException,
-			throws MessageRejectedException, 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;
+			boolean waitForAck = this.waitForAck;
-			if (this.waitForAck) {
+			if (waitForAck) {
 				countdownLatch = new CountDownLatch(ackCounter);
 				this.ackControl.put(messageId, countdownLatch);
 			}
-			packet = this.mapper.fromMessage(message);
+			convertAndSend(message);
-			this.send(packet);
+			if (waitForAck) {
-			if (logger.isDebugEnabled()) {
-				logger.debug("Sent packet for message " + message + " to " + packet.getSocketAddress());
-			}
-			if (this.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 {
+	protected void convertAndSend(Message<?> message) throws Exception {
-		DatagramSocket socket = this.getSocket();
+		DatagramSocket socket;
-		packet.setSocketAddress(this.getDestinationAddress());
+		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 {

spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-4.3.xsd

@@ -80,8 +80,29 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="host" type="xsd:string" />
+					<xsd:attribute name="host" type="xsd:string"/>
+					<xsd:attribute name="destination-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								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.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="acknowledge" type="xsd:string" />
+					<xsd:attribute name="socket-expression" type="xsd:string" use="optional">
+						<xsd:annotation>
+							<xsd:documentation>
+								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).
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="ack-host" type="xsd:string" />
 					<xsd:attribute name="ack-port" type="xsd:string" />
 					<xsd:attribute name="ack-timeout" type="xsd:string" />

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);

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 {

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");
 		}
+
 	}
 
-
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-socket-expression-context.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+
+	<int-ip:udp-inbound-channel-adapter id="inbound" port="0" channel="in" />
+
+	<int:channel id="in" />
+
+	<int:transformer expression="new String(payload).toUpperCase()" input-channel="in" output-channel="out"/>
+
+	<int:channel id="out" />
+
+	<int-ip:udp-outbound-channel-adapter socket-expression="@inbound.socket"
+										 destination-expression="'udp://localhost:' + headers['ip_packetAddress'].port"
+										 channel="out" />
+
+</beans>

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 `<int-ip:udp-outbound-channel-adapter>` (`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]
+----
+<int-ip:udp-inbound-channel-adapter id="inbound" port="0" channel="in" />
+
+<int:channel id="in" />
+
+<int:transformer expression="new String(payload).toUpperCase()"
+                       input-channel="in" output-channel="out"/>
+
+<int:channel id="out" />
+
+<int-ip:udp-outbound-channel-adapter id="outbound"
+                        socket-expression="@inbound.socket"
+                        destination-expression="headers['ip_packetAddress']"
+                        channel="out" />
+----
+
+
 [source,xml]
 ----
 <int-ip:udp-inbound-channel-adapter id="udpReceiver"
@@ -1220,6 +1257,13 @@ For a multicast adapter it is also used to determine which interface the multica
 If not supplied, an internal single threaded executor will be used.
 Needed on some platforms that require the use of specific task executors such as a WorkManagerTaskExecutor.
 One thread will be dedicated to handling acknowledgments (if the acknowledge option is true).
+| destination-expression
+| SpEL expression
+| A SpEL expression to be evaluated to determine which `SocketAddress` to use as a destination address for the
+outgoing UDP packets.
+| socket-expression
+| SpEL expression
+| A SpEL expression to be evaluated to determine which datagram socket use for sending outgoing UDP packets.
 |===
 
 .TCP Inbound Channel Adapter Attributes
@@ -1407,7 +1451,7 @@ For example, when using SSL, properties of the `SSLSession` can be added by obta
 [[ip-annotation]]
 === Annotation-Based Configuration
 
-The following example from the samples respository is used to illustrate some of the configuration options when using
+The following example from the samples repository is used to illustrate some of the configuration options when using
 annotations instead of XML.
 
 [source, java]

src/reference/asciidoc/whats-new.adoc

@@ -43,3 +43,7 @@ for more information.
 
 A new `TcpConnectionServerListeningEvent` is emitted when a server connection factory is started.
 See <<tcp-events>> for more information.
+
+The `destination-expression` and `socket-expression` are now available for the `<int-ip:udp-outbound-channel-adapter>`.
+See <<udp-adapters>> for more information.
+