<int-ip:udp-outbound-channel-adapter id="udpOut"
    host="somehost"
    port="11111"
    multicast="false"
    channel="exampleChannel"/>

A simple UDP outbound channel adapter.

	Tip
	When setting multicast to true, provide the multicast address in the host attribute.

UDP is an efficient, but unreliable protocol. Two attributes are added to improve reliability. When check-length is set to true, the adapter precedes the message data with a length field (4 bytes in network byte order). This enables the receiving side to verify the length of the packet received. If a receiving system uses a buffer that is too short the contain the packet, the packet can be truncated. The length header provides a mechanism to detect this.

Starting with version 4.3, the port can be set 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.

<int-ip:udp-outbound-channel-adapter id="udpOut"
    host="somehost"
    port="11111"
    multicast="false"
    check-length="true"
    channel="exampleChannel"/>

An outbound channel adapter that adds length checking to the datagram packets.

	Tip
	The recipient of the packet must also be configured to expect a length to precede the actual data. For a Spring Integration UDP inbound channel adapter, set its check-length attribute.

The second reliability improvement allows an application-level acknowledgment protocol to be used. The receiver must send an acknowledgment to the sender within a specified time.

<int-ip:udp-outbound-channel-adapter id="udpOut"
    host="somehost"
    port="11111"
    multicast="false"
    check-length="true"
    acknowledge="true"
    ack-host="thishost"
    ack-port="22222"
    ack-timeout="10000"
    channel="exampleChannel"/>

An outbound channel adapter that adds length checking to the datagram packets and waits for an acknowledgment.

	Tip
	Setting acknowledge to true implies the recipient of the packet can interpret the header added to the packet containing acknowledgment data (host and port). Most likely, the recipient will be a Spring Integration inbound channel adapter.

	Tip
	When multicast is true, an additional attribute min-acks-for-success specifies how many acknowledgments must be received within the ack-timeout.

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.

@Bean
@ServiceActivator(inputChannel = "udpOut")
public UnicastSendingMessageHandler handler() {
	return new UnicastSendingMessageHandler("localhost", 11111);
}

(or MulticastSendingChannelAdapter for multicast).

@Bean
public IntegrationFlow udpOutFlow() {
	return IntegrationFlows.from("udpOut")
			.handle(Udp.outboundAdapter("localhost", 1234))
			.get();
}

<int-ip:udp-inbound-channel-adapter id="udpReceiver"
    channel="udpOutChannel"
    port="11111"
    receive-buffer-size="500"
    multicast="false"
    check-length="true"/>

A basic unicast inbound udp channel adapter.

<int-ip:udp-inbound-channel-adapter id="udpReceiver"
    channel="udpOutChannel"
    port="11111"
    receive-buffer-size="500"
    multicast="true"
    multicast-address="225.6.7.8"
    check-length="true"/>

A basic multicast inbound udp channel adapter.

By default, reverse DNS lookups are done on inbound packets to convert IP addresses to hostnames for use in message headers. In environments where DNS is not configured, this can cause delays. This default behavior can be overridden by setting the lookup-host attribute to "false".

@Bean
public UnicastReceivingChannelAdapter udpIn() {
	UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(11111);
	adapter.setOutputChannelName("udpChannel");
	return adapter;
}

@Bean
public IntegrationFlow udpIn() {
	return IntegrationFlows.from(Udp.inboundAdapter(11111))
			.channel("udpChannel")
			.get();
}

Starting with version 5.0.2, a UdpServerListeningEvent is emitted when an inbound adapter is started and has begun listening. This is useful when the adapter is configured to listen on port 0, meaning that the operating system chooses the port. It can also be used instead of polling isListening(), if you need to wait before starting some other process that will connect to the socket.

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 a requestMessage as the root object for the evaluation context. The expression must evaluate to an URI, or String in the URI style (see RFC-2396) or SocketAddress. The inbound 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 the incoming datagram.

With the socket-expression, the Outbound Channel Adapter can use e.g. an 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 NAT. This expression must evaluate to a DatagramSocket. The requestMessage is used as the root object for the evaluation context. The socket-expression parameter cannot be used with parameters multicast and acknowledge.

<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" />

The equivalent configuration using Java DSL Configuration:

@Bean
public IntegrationFlow udpEchoUpcaseServer() {
	return IntegrationFlows.from(Udp.inboundAdapter(11111).id("udpIn"))
			.<byte[], String>transform(p -> new String(p).toUpperCase())
			.handle(Udp.outboundAdapter("headers['ip_packetAddress']")
					.socketExpression("@udpIn.socket"))
			.get();
}

As noted above, TCP sockets can be single-use (one request/response) or shared. Shared sockets do not perform well with outbound gateways, in high-volume environments, because the socket can only process one request/response at a time.

To improve performance, users could use collaborating channel adapters instead of gateways, but that requires application-level message correlation. See Section 32.8, “TCP Message Correlation” for more information.

Spring Integration 2.2 introduced a caching client connection factory, where a pool of shared sockets is used, allowing a gateway to process multiple concurrent requests with a pool of shared connections.

It is now possible to configure a TCP connection factory that supports failover to one or more other servers. When sending a message, the factory will iterate over all its configured factories until either the message can be sent, or no connection can be found. Initially, the first factory in the configured list is used; if a connection subsequently fails the next factory will become the current factory.

<bean id="failCF" class="o.s.i.ip.tcp.connection.FailoverClientConnectionFactory">
    <constructor-arg>
        <list>
            <ref bean="clientFactory1"/>
            <ref bean="clientFactory2"/>
        </list>
    </constructor-arg>
</bean>

	Note
	When using the failover connection factory, the singleUse property must be consistent between the factory itself and the list of factories it is configured to use.

Spring Integration version 5.0 introduced this connection factory. It binds a connection to the calling thread and the same connection is reused each time that thread sends a message. This continues until the connection is closed (by the server or network) or until the thread calls the releaseConnection() method. The connections themselves are provided by another client factory implementation; which must be configured to provide non-shared (single-use) connections so that each thread gets a connection.

Example configuration:

@Bean
public TcpNetClientConnectionFactory cf() {
    TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost",
            Integer.parseInt(System.getProperty(PORT)));
    cf.setSingleUse(true);
    return cf;
}

@Bean
public ThreadAffinityClientConnectionFactory tacf() {
    return new ThreadAffinityClientConnectionFactory(cf());
}

@Bean
@ServiceActivator(inputChannel = "out")
public TcpOutboundGateway outGate() {
    TcpOutboundGateway outGate = new TcpOutboundGateway();
    outGate.setConnectionFactory(tacf());
    outGate.setReplyChannelName("toString");
    return outGate;
}

One goal of the IP Endpoints is to provide communication with systems other than another Spring Integration application. For this reason, only message payloads are sent and received, by default. Since 3.0, headers can be transferred, using JSON, Java serialization, or with custom Serializer s and Deserializer s; see Section 32.8.4, “Transferring Headers” for more information. No message correlation is provided by the framework, except when using the gateways, or collaborating channel adapters on the server side. In the paragraphs below we discuss the various correlation techniques available to applications. In most cases, this requires specific application-level correlation of messages, even when message payloads contain some natural correlation data (such as an order number).

The gateways will automatically correlate messages. However, an outbound gateway should only be used for relatively low-volume use. When the connection factory is configured for a single shared connection to be used for all message pairs (single-use="false"), only one message can be processed at a time. A new message will have to wait until the reply to the previous message has been received. When a connection factory is configured for each new message to use a new connection (single-use="true"), the above restriction does not apply. While this may give higher throughput than a shared connection environment, it comes with the overhead of opening and closing a new connection for each message pair.

Therefore, for high-volume messages, consider using a collaborating pair of channel adapters. However, you will need to provide collaboration logic.

Another solution, introduced in Spring Integration 2.2, is to use a CachingClientConnectionFactory, which allows the use of a pool of shared connections.

To achieve high-volume throughput (avoiding the pitfalls of using gateways as mentioned above) you may consider configuring a pair of collaborating outbound and inbound channel adapters. Collaborating adapters can also be used (server-side or client-side) for totally asynchronous communication (rather than with request/reply semantics). On the server side, message correlation is automatically handled by the adapters because the inbound adapter adds a header allowing the outbound adapter to determine which connection to use to send the reply message.

	Note
	On the server side, care must be taken to populate the ip_connectionId header because it is used to correlate the message to a connection. Messages that originate at the inbound adapter will automatically have the header set. If you wish to construct other messages to send, you will need to set the header. The header value can be captured from an incoming message.

On the client side, the application will have to provide its own correlation logic, if needed. This can be done in a number of ways.

If the message payload has some natural correlation data, such as a transaction id or an order number, AND there is no need to retain any information (such as a reply channel header) from the original outbound message, the correlation is simple and would done at the application level in any case.

If the message payload has some natural correlation data, such as a transaction id or an order number, but there is a need to retain some information (such as a reply channel header) from the original outbound message, you may need to retain a copy of the original outbound message (perhaps by using a publish-subscribe channel) and use an aggregator to recombine the necessary data.

For either of the previous two paragraphs, if the payload has no natural correlation data, you may need to provide a transformer upstream of the outbound channel adapter to enhance the payload with such data. Such a transformer may transform the original payload to a new object containing both the original payload and some subset of the message headers. Of course, live objects (such as reply channels) from the headers can not be included in the transformed payload.

If such a strategy is chosen you will need to ensure the connection factory has an appropriate serializer/deserializer pair to handle such a payload, such as the DefaultSerializer/Deserializer which use java serialization, or a custom serializer and deserializer. The ByteArray*Serializer options mentioned in Section 32.3, “TCP Connection Factories”, including the default ByteArrayCrLfSerializer, do not support such payloads, unless the transformed payload is a String or byte[],

Note

Before the 2.2 release, when a client connection factory was used by collaborating channel adapters, the so-timeout attribute defaulted to the default reply timeout (10 seconds). This meant that if no data were received by the inbound adapter for this period of time, the socket was closed. This default behavior was not appropriate in a truly asynchronous environment, so it now defaults to an infinite timeout. You can reinstate the previous default behavior by setting the so-timeout attribute on the client connection factory to 10000 milliseconds.

TCP is a streaming protocol; Serializers and Deserializers are used to demarcate messages within the stream. Prior to 3.0, only message payloads (String or byte[]) could be transferred over TCP. Beginning with 3.0, you can now transfer selected headers as well as the payload. It is important to understand, though, that "live" objects, such as the replyChannel header cannot be serialized.

Sending header information over TCP requires some additional configuration.

The first step is to provide the ConnectionFactory with a MessageConvertingTcpMessageMapper using the mapper attribute. This mapper delegates to any MessageConverter implementation to convert the message to/from some object that can be (de)serialized by the configured serializer and deserializer.

A MapMessageConverter is provided, which allows the specification of a list of headers that will be added to a Map object, along with the payload. The generated Map has two entries: payload and headers. The headers entry is itself a Map containing the selected headers.

The second step is to provide a (de)serializer that can convert between a Map and some wire format. This can be a custom (de)Serializer, which would typically be needed if the peer system is not a Spring Integration application.

A MapJsonSerializer is provided that will convert a Map to/from JSON. This uses a Spring Integration JsonObjectMapper to perform this function. You can provide a custom JsonObjectMapper if needed. By default, the serializer inserts a linefeed 0x0a character between objects. See the JavaDocs for more information.

	Note
	At the time of writing, the JsonObjectMapper uses whichever version of Jackson is on the classpath.

You can also use standard Java serialization of the Map, using the DefaultSerializer and DefaultDeserializer.

The following example shows the configuration of a connection factory that transfers the correlationId, sequenceNumber, and sequenceSize headers using JSON.

<int-ip:tcp-connection-factory id="client"
    type="client"
    host="localhost"
    port="12345"
    mapper="mapper"
    serializer="jsonSerializer"
    deserializer="jsonSerializer"/>

<bean id="mapper"
      class="o.sf.integration.ip.tcp.connection.MessageConvertingTcpMessageMapper">
    <constructor-arg name="messageConverter">
        <bean class="o.sf.integration.support.converter.MapMessageConverter">
            <property name="headerNames">
                <list>
                    <value>correlationId</value>
                    <value>sequenceNumber</value>
                    <value>sequenceSize</value>
                </list>
            </property>
        </bean>
    </constructor-arg>
</bean>

<bean id="jsonSerializer" class="o.sf.integration.ip.tcp.serializer.MapJsonSerializer" />

A message sent with the above configuration, with payload foo would appear on the wire like so:

{"headers":{"correlationId":"bar","sequenceSize":5,"sequenceNumber":1},"payload":"foo"}

There are some important considerations when using a fixed thread pool with the CallerRunsPolicy (CALLER_RUNS when using the <task/> namespace) and the queue capacity is small.

The following does not apply if you are not using a fixed thread pool.

With NIO connections there are 3 distinct task types; the IO Selector processing is performed on one dedicated thread - detecting events, accepting new connections, and dispatching the IO read operations to other threads, using the task executor. When an IO reader thread (to which the read operation is dispatched) reads data, it hands off to another thread to assemble the incoming message; large messages may take several reads to complete. These "assembler" threads can block waiting for data. When a new read event occurs, the reader determines if this socket already has an assembler and runs a new one if not. When the assembly process is complete, the assembler thread is returned to the pool.

This can cause a deadlock when the pool is exhausted and the CALLER_RUNS rejection policy is in use, and the task queue is full. When the pool is empty and there is no room in the queue, the IO selector thread receives an OP_READ event and dispatches the read using the executor; the queue is full, so the selector thread itself starts the read process; now, it detects that there is not an assembler for this socket and, before it does the read, fires off an assembler; again, the queue is full, and the selector thread becomes the assembler. The assembler is now blocked awaiting the data to be read, which will never happen. The connection factory is now deadlocked because the selector thread can’t handle new events.

We must avoid the selector (or reader) threads performing the assembly task to avoid this deadlock. It is desirable to use seperate pools for the IO and assembly operations.

The framework provides a CompositeExecutor, which allows the configuration of two distinct executors; one for performing IO operations, and one for message assembly. In this environment, an IO thread can never become an assembler thread, and the deadlock cannot occur.

In addition, the task executors should be configured to use a AbortPolicy (ABORT when using <task>). When an IO cannot be completed, it is deferred for a short time and retried continually until it can be completed and an assembler allocated. By default, the delay is 100ms but it can be changed using the readDelay property on the connection factory (read-delay when configuring with the XML namespace).

Example configuration of the composite executor is shown below.

@Bean
private CompositeExecutor compositeExecutor() {
    ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
    ioExec.setCorePoolSize(4);
    ioExec.setMaxPoolSize(10);
    ioExec.setQueueCapacity(0);
    ioExec.setThreadNamePrefix("io-");
    ioExec.setRejectedExecutionHandler(new AbortPolicy());
    ioExec.initialize();
    ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
    assemblerExec.setCorePoolSize(4);
    assemblerExec.setMaxPoolSize(10);
    assemblerExec.setQueueCapacity(0);
    assemblerExec.setThreadNamePrefix("assembler-");
    assemblerExec.setRejectedExecutionHandler(new AbortPolicy());
    assemblerExec.initialize();
    return new CompositeExecutor(ioExec, assemblerExec);
}

<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
    <constructor-arg ref="io"/>
    <constructor-arg ref="assembler"/>
</bean>

<task:executor id="io" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />
<task:executor id="assembler" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />

<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
    <constructor-arg>
        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="threadNamePrefix" value="io-" />
            <property name="corePoolSize" value="4" />
            <property name="maxPoolSize" value="8" />
            <property name="queueCapacity" value="0" />
            <property name="rejectedExecutionHandler">
                <bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
            </property>
        </bean>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="threadNamePrefix" value="assembler-" />
            <property name="corePoolSize" value="4" />
            <property name="maxPoolSize" value="10" />
            <property name="queueCapacity" value="0" />
            <property name="rejectedExecutionHandler">
                <bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
            </property>
        </bean>
    </constructor-arg>
</bean>

Secure Sockets Layer/Transport Layer Security is supported. When using NIO, the JDK 5+ SSLEngine feature is used to handle handshaking after the connection is established. When not using NIO, standard SSLSocketFactory and SSLServerSocketFactory objects are used to create connections. A number of strategy interfaces are provided to allow significant customization; default implementations of these interfaces provide for the simplest way to get started with secure communications.

Regardless of whether NIO is being used, you need to configure the ssl-context-support attribute on the connection factory. This attribute references a <bean/> definition that describes the location and passwords for the required key stores.

SSL/TLS peers require two keystores each; a keystore containing private/public key pairs identifying the peer; a truststore, containing the public keys for peers that are trusted. See the documentation for the keytool utility provided with the JDK. The essential steps are

Repeat for the other peer.

	Note
	It is common in test cases to use the same key stores on both peers, but this should be avoided for production.

After establishing the key stores, the next step is to indicate their locations to the TcpSSLContextSupport bean, and provide a reference to that bean to the connection factory.

<bean id="sslContextSupport"
    class="o.sf.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport">
    <constructor-arg value="client.ks"/>
    <constructor-arg value="client.truststore.ks"/>
    <constructor-arg value="secret"/>
    <constructor-arg value="secret"/>
</bean>

<ip:tcp-connection-factory id="clientFactory"
    type="client"
    host="localhost"
    port="1234"
    ssl-context-support="sslContextSupport" />

The DefaulTcpSSLContextSupport class also has an optional protocol property, which can be SSL or TLS (default).

The keystore file names (first two constructor arguments) use the Spring Resource abstraction; by default the files will be located on the classpath, but this can be overridden by using the file: prefix, to find the files on the filesystem instead.

Starting with version 4.3.6, when using NIO, you can specify an ssl-handshake-timeout (seconds) on the connection factory. This timeout (default 30) is used during SSL handshake when waiting for data; if the timeout is exceeded, the process is aborted and the socket closed.

In many cases, the configuration described above is all that is needed to enable secure communication over TCP/IP. However, a number of strategy interfaces are provided to allow customization and modification of socket factories and sockets.

TcpSSLContextSupport. 

public interface TcpSSLContextSupport {

    SSLContext getSSLContext() throws Exception;

}

Implementations of this interface are responsible for creating an SSLContext. The implementation provided by the framework is the DefaultTcpSSLContextSupport described above. If you require different behavior, implement this interface and provide the connection factory with a reference to a bean of your class' implementation.

TcpSocketFactorySupport. 

public interface TcpSocketFactorySupport {

    ServerSocketFactory getServerSocketFactory();

    SocketFactory getSocketFactory();

}

Implementations of this interface are responsible for obtaining references to ServerSocketFactory and SocketFactory. Two implementations are provided; the first is DefaultTcpNetSocketFactorySupport for non-SSL sockets (when no ssl-context-support attribute is defined); this simply uses the JDK’s default factories. The second implementation is DefaultTcpNetSSLSocketFactorySupport; this is used, by default, when an ssl-context-support attribute is defined; it uses the SSLContext created by that bean to create the socket factories.

	Note
	This interface only applies if using-nio is "false"; socket factories are not used by NIO.

TcpSocketSupport. 

public interface TcpSocketSupport {

    void postProcessServerSocket(ServerSocket serverSocket);

    void postProcessSocket(Socket socket);

}

Implementations of this interface can modify sockets after they are created, and after all configured attributes have been applied, but before the sockets are used. This applies whether or not NIO is being used. For example, you could use an implementation of this interface to modify the supported cipher suites on an SSL socket, or you could add a listener that gets notified after SSL handshaking is complete. The sole implementation provided by the framework is the DefaultTcpSocketSupport which does not modify the sockets in any way

To supply your own implementation of TcpSocketFactorySupport or TcpSocketSupport, provide the connection factory with references to beans of your custom type using the socket-factory-support and socket-support attributes, respectively.

TcpNetConnectionSupport. 

public interface TcpNetConnectionSupport {

	TcpNetConnection createNewConnection(Socket socket,
			boolean server, boolean lookupHost,
			ApplicationEventPublisher applicationEventPublisher,
			String connectionFactoryName) throws Exception;

}

This interface is invoked to create TcpNetConnection objects (or objects from subclasses). The framework provides a single implementation DefatulTcpNetConnectionSupport which creates simple TcpNetConnection objects by default. It has two properties pushbackCapable and pushbackBufferSize; when push back is enabled, the implementation returns a subclass that wraps the connection’s InputStream in a PushbackInputStream. Aligned with the PushbackInputStream default, the buffer size defaults to 1. This enables deserializers to "unread" (push back) bytes into the stream. The following is a trivial example of how it might be used in a delegating deserializer which "peeks" at the first byte to determine which deserializer to invoke:

public class CompositeDeserializer implements Deserializer<byte[]> {

    private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer();

    private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer();

    @Override
    public byte[] deserialize(InputStream inputStream) throws IOException {
        PushbackInputStream pbis = (PushbackInputStream) inputStream;
        int first = pbis.read();
        if (first < 0) {
            throw new SoftEndOfStreamException();
        }
        pbis.unread(first);
        if (first == ByteArrayStxEtxSerializer.STX) {
            this.receivedStxEtx = true;
            return this.stxEtx.deserialize(pbis);
        }
        else {
            this.receivedCrLf = true;
            return this.crlf.deserialize(pbis);
        }
    }

}

TcpNioConnectionSupport. 

public interface TcpNioConnectionSupport {

    TcpNioConnection createNewConnection(SocketChannel socketChannel,
            boolean server, boolean lookupHost,
            ApplicationEventPublisher applicationEventPublisher,
            String connectionFactoryName) throws Exception;

}

This interface is invoked to create TcpNioConnection objects (or objects from subclasses). Two implementations are provided DefaultTcpNioSSLConnectionSupport and DefaultTcpNioConnectionSupport which are used depending on whether SSL is in use or not. A common use case would be to subclass DefaultTcpNioSSLConnectionSupport and override postProcessSSLEngine; see the example below. As with the DefatulTcpNetConnectionSupport, these implementations also support push back.

To enable client certificate authentication when using SSL, the technique depends on whether NIO is in use or not. When NIO is not being used, provide a custom TcpSocketSupport implementation to post-process the server socket:

serverFactory.setTcpSocketSupport(new DefaultTcpSocketSupport() {

    @Override
    public void postProcessServerSocket(ServerSocket serverSocket) {
        ((SSLServerSocket) serverSocket).setNeedClientAuth(true);
    }

});

(When using XML configuration, provide a reference to your bean using the socket-support attribute).

When using NIO, provide a custom TcpNioSslConnectionSupport implementation to post-process the SSLEngine.

@Bean
public DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport() {
    return new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport) {

            @Override
            protected void postProcessSSLEngine(SSLEngine sslEngine) {
                sslEngine.setNeedClientAuth(true);
            }

    }
}

@Bean
public TcpNioServerConnectionFactory server() {
    ...
    serverFactory.setTcpNioConnectionSupport(tcpNioConnectionSupport());
    ...
}

(When using XML configuration, since version 4.3.7, provide a reference to your bean using the nio-connection-support attribute).