From a79be71ed2eafcaf0b1cf35dbdf7c51e976070ab Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 22 Oct 2020 13:32:56 -0400 Subject: [PATCH] GH-3410: Add UDP SocketCustomizer Resolves https://github.com/spring-projects/spring-integration/issues/3410 **cherry-pick to 5.3.x** * Doc polishing. --- .../ip/config/IpAdapterParserUtils.java | 4 +- .../UdpInboundChannelAdapterParser.java | 4 +- .../UdpOutboundChannelAdapterParser.java | 3 ++ ...AbstractUdpOutboundChannelAdapterSpec.java | 12 +++++ .../ip/dsl/UdpInboundChannelAdapterSpec.java | 34 +++++++++----- .../integration/ip/dsl/package-info.java | 1 + .../integration/ip/udp/SocketCustomizer.java | 38 +++++++++++++++ .../udp/UnicastReceivingChannelAdapter.java | 16 ++++++- .../ip/udp/UnicastSendingMessageHandler.java | 31 ++++++++---- .../ip/config/spring-integration-ip.xsd | 12 +++++ .../ip/config/ParserTests-context.xml | 4 ++ .../ip/config/ParserUnitTests.java | 16 +++++++ .../ip/dsl/IpIntegrationTests.java | 10 +++- src/reference/asciidoc/ip.adoc | 47 +++++++++++-------- src/reference/asciidoc/whats-new.adoc | 5 +- 15 files changed, 192 insertions(+), 45 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/SocketCustomizer.java 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 8adaacd46b..bd358047aa 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-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() { } 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 d7ec431b79..4f1637016f 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-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 @@ public class UdpInboundChannelAdapterParser extends AbstractChannelAdapterParser IpAdapterParserUtils.TASK_EXECUTOR); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.LOOKUP_HOST); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.UDP_SOCKET_CUSTOMIZER); return builder.getBeanDefinition(); } 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 684efdf3c6..8352708bb5 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 @@ -36,6 +36,7 @@ import org.springframework.util.StringUtils; */ public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = parseUdp(element, parserContext); IpAdapterParserUtils.addCommonSocketOptions(builder, element); @@ -86,6 +87,8 @@ public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap IpAdapterParserUtils.TASK_EXECUTOR); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "socket-expression", "socketExpressionString"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.UDP_SOCKET_CUSTOMIZER); return builder; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java index e3aa00c6e8..6024c8ea45 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java @@ -21,6 +21,7 @@ import java.util.function.Function; 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 abstract class AbstractUdpOutboundChannelAdapterSpec { }; /** * Constructs a UnicastReceivingChannelAdapter that listens on the specified port. @@ -89,6 +91,16 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece 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 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece if (this.soSendBufferSize > 0) { out.setSendBufferSize(this.soSendBufferSize); } + this.socketCustomizer.configure(out); out.send(ackPack); out.close(); } @@ -258,7 +271,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece } /** - * 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. @@ -269,6 +282,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece if (soReceiveBufferSize > 0) { socket.setReceiveBufferSize(soReceiveBufferSize); } + this.socketCustomizer.configure(socket); } @Override 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 dd4ca5765f..ddc2d516fe 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 @@ -101,6 +101,8 @@ public class UnicastSendingMessageHandler extends private EvaluationContext evaluationContext; + private SocketCustomizer socketCustomizer = socket -> { }; + private volatile CountDownLatch ackLatch; private volatile boolean ackThreadRunning; @@ -206,6 +208,25 @@ public class UnicastSendingMessageHandler extends 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) { @@ -224,15 +245,6 @@ public class UnicastSendingMessageHandler extends } } - /** - * @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) { @@ -495,6 +507,7 @@ public class UnicastSendingMessageHandler extends if (soSendBufferSize > 0) { socket.setSendBufferSize(soSendBufferSize); } + this.socketCustomizer.configure(socket); } /** diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd index 3c033e0155..23b0f551d4 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip.xsd @@ -873,6 +873,18 @@ + + + + A SocketCustomizer bean. + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml index 0688fe2d5b..8989575728 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserTests-context.xml @@ -19,6 +19,8 @@ + + received = this.udpIn.receive(10000); assertThat(received).isNotNull(); assertThat(Transformers.objectToString().transform(received).getPayload()).isEqualTo("foo"); + assertThat(TestUtils.getPropertyValue(this.udpOutbound, "socket", DatagramSocket.class).getTrafficClass()) + .isEqualTo(0x10); } @Test @@ -303,7 +310,8 @@ public class IpIntegrationTests { @Bean public IntegrationFlow outUdpAdapter() { - return f -> f.handle(Udp.outboundAdapter(m -> m.getHeaders().get("udp_dest"))); + return f -> f.handle(Udp.outboundAdapter(m -> m.getHeaders().get("udp_dest")) + .configureSocket(socket -> socket.setTrafficClass(0x10))); } @Bean diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 672972ab34..a0d78011ef 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -72,6 +72,7 @@ The following example configures a UDP outbound channel adapter: host="somehost" port="11111" multicast="false" + socket-customizer="udpCustomizer" channel="exampleChannel"/> ---- ==== @@ -88,7 +89,9 @@ The `length` header provides a mechanism to detect this. Starting with version 4.3, you can set the `port` to `0`, in which case the operating system chooses the port. The chosen port can be discovered by invoking `getPort()` after the adapter is started and `isListening()` returns `true`. -The preceding example shows an outbound channel adapter that adds length checking to the datagram packets: +Starting with version 5.3.3, you can add a `SocketCustomizer` bean to modify the `DatagramSocket` after it is created (for example, call `setTrafficClass(0x10)`). + +The following example shows an outbound channel adapter that adds length checking to the datagram packets: ==== [source,xml] @@ -143,7 +146,7 @@ The following example shows how to configure an outbound UDP adapter with Java: @Bean @ServiceActivator(inputChannel = "udpOut") public UnicastSendingMessageHandler handler() { - return new UnicastSendingMessageHandler("localhost", 11111); + return new UnicastSendingMessageHandler("localhost", 11111); } ---- ==== @@ -159,9 +162,9 @@ The following example shows how to configure an outbound UDP adapter with the Ja ---- @Bean public IntegrationFlow udpOutFlow() { - return IntegrationFlows.from("udpOut") - .handle(Udp.outboundAdapter("localhost", 1234)) - .get(); + return f -> f.handle(Udp.outboundAdapter("localhost", 1234) + .configureSocket(socket -> socket.setTrafficClass(0x10))) + .get(); } ---- ==== @@ -178,6 +181,7 @@ The following example shows how to configure a basic unicast inbound udp channel port="11111" receive-buffer-size="500" multicast="false" + socket-customizer="udpCustomizer" check-length="true"/> ---- ==== @@ -201,6 +205,9 @@ By default, reverse DNS lookups are done on inbound packets to convert IP addres In environments where DNS is not configured, this can cause delays. You can override this default behavior by setting the `lookup-host` attribute to `false`. +Starting with version 5.3.3, you can add a `SocketCustomizer` bean to modify the `DatagramSocket` after it is created. +It is called for the receiving socket and any sockets created for sending acks. + ==== Inbound UDP Adapters (Java Configuration) The following example shows how to configure an inbound UDP adapter with Java: @@ -209,9 +216,9 @@ The following example shows how to configure an inbound UDP adapter with Java: ---- @Bean public UnicastReceivingChannelAdapter udpIn() { - UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(11111); - adapter.setOutputChannelName("udpChannel"); - return adapter; + UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(11111); + adapter.setOutputChannelName("udpChannel"); + return adapter; } ---- @@ -223,9 +230,9 @@ The following example shows how to configure an inbound UDP adapter with the Jav ---- @Bean public IntegrationFlow udpIn() { - return IntegrationFlows.from(Udp.inboundAdapter(11111)) - .channel("udpChannel") - .get(); + return IntegrationFlows.from(Udp.inboundAdapter(11111)) + .channel("udpChannel") + .get(); } ---- @@ -278,11 +285,11 @@ The following example shows the equivalent configuration with the Java DSL: ---- @Bean public IntegrationFlow udpEchoUpcaseServer() { - return IntegrationFlows.from(Udp.inboundAdapter(11111).id("udpIn")) - .transform(p -> new String(p).toUpperCase()) - .handle(Udp.outboundAdapter("headers['ip_packetAddress']") - .socketExpression("@udpIn.socket")) - .get(); + return IntegrationFlows.from(Udp.inboundAdapter(11111).id("udpIn")) + .transform(p -> new String(p).toUpperCase()) + .handle(Udp.outboundAdapter("headers['ip_packetAddress']") + .socketExpression("@udpIn.socket")) + .get(); } ---- ==== @@ -1396,10 +1403,10 @@ The following listing shows the definition of the `TcpNetConnectionSupport` stra ---- public interface TcpNetConnectionSupport { - TcpNetConnection createNewConnection(Socket socket, - boolean server, boolean lookupHost, - ApplicationEventPublisher applicationEventPublisher, - String connectionFactoryName) throws Exception; + TcpNetConnection createNewConnection(Socket socket, + boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, + String connectionFactoryName) throws Exception; } ---- diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e57ecb66b9..8c36b3a804 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -67,13 +67,16 @@ The legacy metrics that were replaced by Micrometer meters have been removed. The <<./barrier.adoc#barrier,Thread Barrier>> has now two separate timeout options: `requestTimeout` and `triggerTimeout`. [[x5.4-tcp]] -=== TCP Changes +=== TCP/UDP Changes Connection factories now support multiple sending components (`TcpSender`); they remain limited to one receiving component (`TcpListener`). This allows, for example, an inbound gateway and outbound channel adapter to share the same factory, supporting both request/reply and arbitrary messaging from the server to the client. Shared factories should not be used with outbound gateways, unless single-use connections or the `ThreadAffinityClientConnectionFactory` are being used. See <<./ip.adoc#ip-collaborating-adapters,Collaborating Channel Adapters>> and <<./ip.adoc#tcp-gateways, TCP Gateways>> for more information. +The UDP channel adapters can now be configured with a `SocketCustomizer` which allows the setting of socket properties that are not directly supported by the adapters. +See <<./ip.adoc#udp-adapters,UDP Adapters>> for more information. + [[x5.4-rmi]] === RMI Changes