GH-3410: Add UDP SocketCustomizer

Resolves #3410

**cherry-pick to 5.3.x**

* Doc polishing.
# Conflicts:
#	spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
#	spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java
#	src/reference/asciidoc/whats-new.adoc

Author: garyrussell

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -133,6 +133,8 @@ public abstract class IpAdapterParserUtils {
 
 	public static final String CONNECT_TIMEOUT = "connect-timeout";
 
+	public static final String UDP_SOCKET_CUSTOMIZER = "socket-customizer";
+
 	private IpAdapterParserUtils() {
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -51,6 +51,8 @@ protected AbstractBeanDefinition doParse(Element element, ParserContext parserCo
 				IpAdapterParserUtils.TASK_EXECUTOR);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.LOOKUP_HOST);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.UDP_SOCKET_CUSTOMIZER);
 		return builder.getBeanDefinition();
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpOutboundChannelAdapterParser.java

@@ -35,6 +35,7 @@
  */
 public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
 
+	@Override
 	protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = parseUdp(element, parserContext);
 		IpAdapterParserUtils.addCommonSocketOptions(builder, element);
@@ -84,6 +85,8 @@ private BeanDefinitionBuilder parseUdp(Element element, ParserContext parserCont
 				IpAdapterParserUtils.TASK_EXECUTOR);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				"socket-expression", "socketExpressionString");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.UDP_SOCKET_CUSTOMIZER);
 		return builder;
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java

@@ -21,6 +21,7 @@
 
 import org.springframework.integration.dsl.MessageHandlerSpec;
 import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.ip.udp.SocketCustomizer;
 import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
 import org.springframework.messaging.Message;
 
@@ -133,4 +134,15 @@ public S socketExpression(String socketExpression) {
 		return _this();
 	}
 
+	/**
+	 * Configure the socket.
+	 * @param customizer the customizer.
+	 * @return the spec.
+	 * @since 5.3.3
+	 */
+	public S configureSocket(SocketCustomizer customizer) {
+		this.target.setSocketCustomizer(customizer);
+		return _this();
+	}
+
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpInboundChannelAdapterSpec.java

@@ -21,6 +21,7 @@
 
 import org.springframework.integration.dsl.MessageProducerSpec;
 import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
+import org.springframework.integration.ip.udp.SocketCustomizer;
 import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
 import org.springframework.scheduling.TaskScheduler;
 
@@ -50,7 +51,7 @@ protected UdpInboundChannelAdapterSpec(int port, String multicastGroup) {
 	 */
 	public UdpInboundChannelAdapterSpec soTimeout(int soTimeout) {
 		this.target.setSoTimeout(soTimeout);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -60,7 +61,7 @@ public UdpInboundChannelAdapterSpec soTimeout(int soTimeout) {
 	 */
 	public UdpInboundChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler) {
 		this.target.setTaskScheduler(taskScheduler);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -70,7 +71,7 @@ public UdpInboundChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler) {
 	 */
 	public UdpInboundChannelAdapterSpec soReceiveBufferSize(int soReceiveBufferSize) {
 		this.target.setSoReceiveBufferSize(soReceiveBufferSize);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -80,7 +81,7 @@ public UdpInboundChannelAdapterSpec soReceiveBufferSize(int soReceiveBufferSize)
 	 */
 	public UdpInboundChannelAdapterSpec receiveBufferSize(int receiveBufferSize) {
 		this.target.setReceiveBufferSize(receiveBufferSize);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -90,7 +91,7 @@ public UdpInboundChannelAdapterSpec receiveBufferSize(int receiveBufferSize) {
 	 */
 	public UdpInboundChannelAdapterSpec lengthCheck(boolean lengthCheck) {
 		this.target.setLengthCheck(lengthCheck);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -100,7 +101,7 @@ public UdpInboundChannelAdapterSpec lengthCheck(boolean lengthCheck) {
 	 */
 	public UdpInboundChannelAdapterSpec localAddress(String localAddress) {
 		this.target.setLocalAddress(localAddress);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -110,7 +111,7 @@ public UdpInboundChannelAdapterSpec localAddress(String localAddress) {
 	 */
 	public UdpInboundChannelAdapterSpec poolSize(int poolSize) {
 		this.target.setPoolSize(poolSize);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -120,7 +121,7 @@ public UdpInboundChannelAdapterSpec poolSize(int poolSize) {
 	 */
 	public UdpInboundChannelAdapterSpec taskExecutor(Executor taskExecutor) {
 		this.target.setTaskExecutor(taskExecutor);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -130,7 +131,7 @@ public UdpInboundChannelAdapterSpec taskExecutor(Executor taskExecutor) {
 	 */
 	public UdpInboundChannelAdapterSpec socket(DatagramSocket socket) {
 		this.target.setSocket(socket);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -140,7 +141,7 @@ public UdpInboundChannelAdapterSpec socket(DatagramSocket socket) {
 	 */
 	public UdpInboundChannelAdapterSpec soSendBufferSize(int soSendBufferSize) {
 		this.target.setSoSendBufferSize(soSendBufferSize);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -150,7 +151,18 @@ public UdpInboundChannelAdapterSpec soSendBufferSize(int soSendBufferSize) {
 	 */
 	public UdpInboundChannelAdapterSpec lookupHost(boolean lookupHost) {
 		this.target.setLookupHost(lookupHost);
-		return _this();
+		return this;
+	}
+
+	/**
+	 * Configure the socket.
+	 * @param customizer the customizer.
+	 * @return the spec.
+	 * @since 5.3.3
+	 */
+	public UdpInboundChannelAdapterSpec configureSocket(SocketCustomizer customizer) {
+		this.target.setSocketCustomizer(customizer);
+		return this;
 	}
 
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/package-info.java

@@ -1,4 +1,5 @@
 /**
  * Provides TCP/UDP Component support for the Java DSL.
  */
+@org.springframework.lang.NonNullApi
 package org.springframework.integration.ip.dsl;

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.SocketException;
+
+/**
+ * Configures a socket.
+ *
+ * @author Gary Russell
+ * @since 5.3.3
+ */
+@FunctionalInterface
+public interface SocketCustomizer {
+
+	/**
+	 * Configure the socket ({code setTrafficClass()}, etc).
+	 * @param socket the socket.
+	 * @throws SocketException a socket exception.
+	 */
+	void configure(DatagramSocket socket) throws SocketException;
+
+}

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

@@ -35,6 +35,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
+import org.springframework.util.Assert;
 
 /**
  * A channel adapter to receive incoming UDP packets. Packets can optionally be preceded by a
@@ -54,6 +55,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
 
 	private static Pattern addressPattern = Pattern.compile("([^:]*):([0-9]*)");
 
+	private SocketCustomizer socketCustomizer = socket -> { };
 
 	/**
 	 * Constructs a UnicastReceivingChannelAdapter that listens on the specified port.
@@ -85,6 +87,16 @@ public void setLengthCheck(boolean lengthCheck) {
 		this.mapper.setLengthCheck(lengthCheck);
 	}
 
+	/**
+	 * Set a customizer to further configure the socket after creation.
+	 * @param socketCustomizer the customizer.
+	 * @since 5.3.3
+	 */
+	public void setSocketCustomizer(SocketCustomizer socketCustomizer) {
+		Assert.notNull(socketCustomizer, "'socketCustomizer' cannot be null");
+		this.socketCustomizer = socketCustomizer;
+	}
+
 	@Override
 	public boolean isLongLived() {
 		return true;
@@ -169,6 +181,7 @@ protected void sendAck(Message<byte[]> message) {
 			if (this.soSendBufferSize > 0) {
 				out.setSendBufferSize(this.soSendBufferSize);
 			}
+			this.socketCustomizer.configure(out);
 			out.send(ackPack);
 			out.close();
 		}
@@ -260,7 +273,7 @@ public synchronized DatagramSocket getSocket() {
 	}
 
 	/**
-	 * Sets timeout and receive buffer size
+	 * Sets timeout and receive buffer size; calls the socket customizer.
 	 *
 	 * @param socket The socket.
 	 * @throws SocketException Any socket exception.
@@ -272,6 +285,7 @@ protected void setSocketAttributes(DatagramSocket socket)
 		if (soReceiveBufferSize > 0) {
 			socket.setReceiveBufferSize(soReceiveBufferSize);
 		}
+		this.socketCustomizer.configure(socket);
 	}
 
 	@Override

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

@@ -105,6 +105,9 @@ public class UnicastSendingMessageHandler extends
 
 	private EvaluationContext evaluationContext;
 
+	private SocketCustomizer socketCustomizer = socket -> { };
+
+
 	/**
 	 * Basic constructor; no reliability; no acknowledgment.
 	 * @param host Destination host.
@@ -204,6 +207,25 @@ public UnicastSendingMessageHandler(String host,
 				ackTimeout);
 	}
 
+	/**
+	 * @param lengthCheck if true, a four byte binary length header is added to the
+	 * packet, allowing the receiver to check for data truncation.
+	 * @since 5.0
+	 */
+	public void setLengthCheck(boolean lengthCheck) {
+		this.mapper.setLengthCheck(lengthCheck);
+	}
+
+	/**
+	 * Set a customizer to further configure the socket after creation.
+	 * @param socketCustomizer the customizer.
+	 * @since 5.3.3
+	 */
+	public void setSocketCustomizer(SocketCustomizer socketCustomizer) {
+		Assert.notNull(socketCustomizer, "'socketCustomizer' cannot be null");
+		this.socketCustomizer = socketCustomizer;
+	}
+
 	protected final void setReliabilityAttributes(boolean lengthCheck,
 			boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
 		this.mapper.setLengthCheck(lengthCheck);
@@ -221,15 +243,6 @@ protected final void setReliabilityAttributes(boolean lengthCheck,
 		}
 	}
 
-	/**
-	 * @param lengthCheck if true, a four byte binary length header is added to the
-	 * packet, allowing the receiver to check for data truncation.
-	 * @since 5.0
-	 */
-	public void setLengthCheck(boolean lengthCheck) {
-		this.mapper.setLengthCheck(lengthCheck);
-	}
-
 	@Override
 	public void doStart() {
 		if (this.acknowledge) {
@@ -496,6 +509,7 @@ protected void setSocketAttributes(DatagramSocket socket) throws SocketException
 		if (this.getSoSendBufferSize() > 0) {
 			socket.setSendBufferSize(this.getSoSendBufferSize());
 		}
+		this.socketCustomizer.configure(socket);
 	}
 
 	/**

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

@@ -873,6 +873,18 @@
 			<xsd:extension base="common-attributes">
 				<xsd:attribute name="check-length" type="xsd:string"/>
 				<xsd:attribute name="multicast" type="xsd:string"/>
+				<xsd:attribute name="socket-customizer" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							A SocketCustomizer bean.
+						</xsd:documentation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.integration.ip.udp.SocketCustomizer"/>
+							</tool:annotation>
+						</xsd:appinfo>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>