The following example configures a UDP outbound channel adapter:

UDP is an efficient but unreliable protocol. Spring Integration adds two attributes to improve reliability: check-length and acknowledge. When check-length is set to true, the adapter precedes the message data with a length field (four 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 to contain the packet, the packet can be truncated. 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.

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:

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.

The following example shows an outbound channel adapter that adds length checking to the datagram packets and waits for an acknowledgment:

Starting with version 4.3, you can set the ackPort to 0, in which case the operating system chooses the port.

The following example shows how to configure an outbound UDP adapter with Java:

(or MulticastSendingChannelAdapter for multicast).

The following example shows how to configure an outbound UDP adapter with the Java DSL:

The following example shows how to configure a basic unicast inbound udp channel adapter.

The following example shows how to configure a basic multicast inbound udp channel adapter:

By default, reverse DNS lookups are done on inbound packets to convert IP addresses to host names for use in message headers. 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.

The following example shows how to configure an inbound UDP adapter with Java:

The following example shows how to configure an inbound UDP adapter with the Java DSL:

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 <int-ip:udp-outbound-channel-adapter> (UnicastSendingMessageHandler) has destination-expression and socket-expression options.

You can use the destination-expression as a runtime alternative to the hardcoded host-port pair to determine the destination address for the outgoing datagram packet against a requestMessage (with the root object for the evaluation context). The expression must evaluate to an URI, a String in the URI style (see RFC-2396), or a SocketAddress. You can also use the inbound IpHeaders.PACKET_ADDRESS header for this expression. In the framework, the DatagramPacketMessageMapper populates this header 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 (for example) an inbound channel adapter socket to send datagrams through the same port which they were received. It is useful in a scenario where our application works as a UDP server and clients operate behind network address translation (NAT). This expression must evaluate to a DatagramSocket. The requestMessage is used as the root object for the evaluation context. You cannot use the socket-expression parameter with the multicast and acknowledge parameters. The following example shows how to configure a UDP inbound channel adapter with a transformer that converts to upper case and uses a socket:

The following example shows the equivalent configuration with the Java DSL:

For TCP, the configuration of the underlying connection is provided by using a connection factory. Two types of connection factory are provided: a client connection factory and a server connection factory. Client connection factories establish outgoing connections. Server connection factories listen for incoming connections.

An outbound channel adapter uses a client connection factory, but you can also provide a reference to a client connection factory to an inbound channel adapter. That adapter receives any incoming messages that are received on connections created by the outbound adapter.

An inbound channel adapter or gateway uses a server connection factory. (In fact, the connection factory cannot function without one). You can also provide a reference to a server connection factory to an outbound adapter. You can then use that adapter to send replies to incoming messages on the same connection.

You may give a reference to a connection factory to a maximum of one adapter of each type.

Spring Integration provides connection factories that use java.net.Socket and java.nio.channel.SocketChannel.

The following example shows a simple server connection factory that uses java.net.Socket connections:

The following example shows a simple server connection factory that uses java.nio.channel.SocketChannel connections:

The following example shows a client connection factory that uses java.net.Socket connections and creates a new connection for each message:

Starting with version 5.2, the client connection factories support the property connectTimeout, specified in seconds, which defaults to 60.

TCP is a streaming protocol. This means that some structure has to be provided to data transported over TCP so that the receiver can demarcate the data into discrete messages. Connection factories are configured to use serializers and deserializers to convert between the message payload and the bits that are sent over TCP. This is accomplished by providing a deserializer and a serializer for inbound and outbound messages, respectively. Spring Integration provides a number of standard serializers and deserializers.

ByteArrayCrlfSerializer^* converts a byte array to a stream of bytes followed by carriage return and linefeed characters (\r\n). This is the default serializer (and deserializer) and can be used (for example) with telnet as a client.

The ByteArraySingleTerminatorSerializer^* converts a byte array to a stream of bytes followed by a single termination character (the default is 0x00).

The ByteArrayLfSerializer^* converts a byte array to a stream of bytes followed by a single linefeed character (0x0a).

The ByteArrayStxEtxSerializer^* converts a byte array to a stream of bytes preceded by an STX (0x02) and followed by an ETX (0x03).

The ByteArrayLengthHeaderSerializer converts a byte array to a stream of bytes preceded by a binary length in network byte order (big endian). This an efficient deserializer because it does not have to parse every byte to look for a termination character sequence. It can also be used for payloads that contain binary data. The preceding serializers support only text in the payload. The default size of the length header is four bytes (an Integer), allowing for messages up to (2^31 - 1) bytes. However, the length header can be a single byte (unsigned) for messages up to 255 bytes, or an unsigned short (2 bytes) for messages up to (2^16 - 1) bytes. If you need any other format for the header, you can subclass ByteArrayLengthHeaderSerializer and provide implementations for the readHeader and writeHeader methods. The absolute maximum data size is (2^31 - 1) bytes. Starting with version 5.2, the header value can include the length of the header in addition to the payload. Set the inclusive property to enable that mechanism (it must be set to the same for producers and consumers).

The ByteArrayRawSerializer^*, converts a byte array to a stream of bytes and adds no additional message demarcation data. With this serializer (and deserializer), the end of a message is indicated by the client closing the socket in an orderly fashion. When using this serializer, message reception hangs until the client closes the socket or a timeout occurs. A timeout does not result in a message. When this serializer is being used and the client is a Spring Integration application, the client must use a connection factory that is configured with single-use="true". Doing so causes the adapter to close the socket after sending the message. The serializer does not, by itself, close the connection. You should use this serializer only with the connection factories used by channel adapters (not gateways), and the connection factories should be used by either an inbound or outbound adapter but not both. See also ByteArrayElasticRawDeserializer, later in this section. However, since version 5.2, the outbound gateway has a new property closeStreamAfterSend; this allows the use of raw serializers/deserializers because the EOF is signaled to the server, while leaving the connection open to receive the reply.

Each of these is a subclass of AbstractByteArraySerializer, which implements both org.springframework.core.serializer.Serializer and org.springframework.core.serializer.Deserializer. For backwards compatibility, connections that use any subclass of AbstractByteArraySerializer for serialization also accept a String that is first converted to a byte array. Each of these serializers and deserializers converts an input stream that contains the corresponding format to a byte array payload.

To avoid memory exhaustion due to a badly behaved client (one that does not adhere to the protocol of the configured serializer), these serializers impose a maximum message size. If an incoming message exceeds this size, an exception is thrown. The default maximum message size is 2048 bytes. You can increase it by setting the maxMessageSize property. If you use the default serializer or deserializer and wish to increase the maximum message size, you must declare the maximum message size as an explicit bean with the maxMessageSize property set and configure the connection factory to use that bean.

The classes marked with ^* earlier in this section use an intermediate buffer and copy the decoded data to a final buffer of the correct size. Starting with version 4.3, you can configure these buffers by setting a poolSize property to let these raw buffers be reused instead of being allocated and discarded for each message, which is the default behavior. Setting the property to a negative value creates a pool that has no bounds. If the pool is bounded, you can also set the poolWaitTimeout property (in milliseconds), after which an exception is thrown if no buffer becomes available. It defaults to infinity. Such an exception causes the socket to be closed.

If you wish to use the same mechanism in custom deserializers, you can extend AbstractPooledBufferByteArraySerializer (instead of its super class, AbstractByteArraySerializer) and implement doDeserialize() instead of deserialize(). The buffer is automatically returned to the pool. AbstractPooledBufferByteArraySerializer also provides a convenient utility method: copyToSizedArray().

Version 5.0 added the ByteArrayElasticRawDeserializer. This is similar to the deserializer side of ByteArrayRawSerializer above, except that it is not necessary to set a maxMessageSize. Internally, it uses a ByteArrayOutputStream that lets the buffer grow as needed. The client must close the socket in an orderly manner to signal end of message.

The MapJsonSerializer uses a Jackson ObjectMapper to convert between a Map and JSON. You can use this serializer in conjunction with a MessageConvertingTcpMessageMapper and a MapMessageConverter to transfer selected headers and the payload in JSON.

The final standard serializer is org.springframework.core.serializer.DefaultSerializer, which you can use to convert serializable objects with Java serialization. org.springframework.core.serializer.DefaultDeserializer is provided for inbound deserialization of streams that contain serializable objects.

If you do not wish to use the default serializer and deserializer (ByteArrayCrLfSerializer), you must set the serializer and deserializer attributes on the connection factory. The following example shows how to do so:

A server connection factory that uses java.net.Socket connections and uses Java serialization on the wire.

For full details of the attributes available on connection factories, see the reference at the end of this section.

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

If your data is not in a format supported by one of the standard deserializers, you can implement your own; you can also implement a custom serializer.

To implement a custom serializer and deserializer pair, implement the org.springframework.core.serializer.Deserializer and org.springframework.core.serializer.Serializer interfaces.

When the deserializer detects a closed input stream between messages, it must throw a SoftEndOfStreamException; this is a signal to the framework to indicate that the close was "normal". If the stream is closed while decoding a message, some other exception should be thrown instead.

Starting with version 5.2, SoftEndOfStreamException is now a RuntimeException instead of extending IOException.

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

To improve performance, you can use collaborating channel adapters instead of gateways, but that requires application-level message correlation. See TCP Message Correlation for more information.

Spring Integration 2.2 introduced a caching client connection factory, which uses a pool of shared sockets, letting a gateway process multiple concurrent requests with a pool of shared connections.

You can configure a TCP connection factory that supports failover to one or more other servers. When sending a message, the factory iterates 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 becomes the current factory. The following example shows how to configure a failover client connection factory:

The connection factory has two properties related to failing back, when used with a shared connection (singleUse=false):

Consider the following scenario based on the above configuration: Let’s say clientFactory1 cannot establish a connection but clientFactory2 can. When the failCF getConnection() method is called after the refreshSharedInterval has passed, we will again attempt to connect using clientFactory1; if successful, the connection to clientFactory2 will be closed. If closeOnRefresh is false, the "old" connection will remain open and may be reused in future if the first factory fails once more.

Set refreshSharedInterval to only attempt to reconnect with the first factory after that time has expired; setting it to Long.MAX_VALUE (default) if you only want to fail back to the first factory when the current connection fails.

Set closeOnRefresh to close the "old" connection after a refresh actually creates a new connection.

Starting with version 5.3, these default to Long.MAX_VALUE and true so the factory only attempts to fail back when the current connection fails. To revert to the default behavior of previous versions, set them to 0 and false.

Also see Testing Connections.

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 the 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.

The following example shows how to configure a TCP thread affinity connection factory:

Gateways automatically correlate messages. However, you should use an outbound gateway for relatively low-volume applications. When you configure the connection factory to use a single shared connection for all message pairs ('single-use="false"'), only one message can be processed at a time. A new message has 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"'), this restriction does not apply. While this setting can 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, to do so, you 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 earlier) you can configure a pair of collaborating outbound and inbound channel adapters. You can also use collaborating adapters (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 that allows the outbound adapter to determine which connection to use when sending the reply message.

On the client side, the application must provide its own correlation logic, if needed. You can do so in a number of ways.

If the message payload has some natural correlation data (such as a transaction ID or an order number) and you have no need to retain any information (such as a reply channel header) from the original outbound message, the correlation is simple and would be 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 you need to retain some information (such as a reply channel header) from the original outbound message, you can 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 scenarios, if the payload has no natural correlation data, you can 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 that contains both the original payload and some subset of the message headers. Of course, live objects (such as reply channels) from the headers cannot be included in the transformed payload.

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

TCP is a streaming protocol. Serializers and Deserializers 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 transfer selected headers as well as the payload. However, “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 that uses the mapper attribute. This mapper delegates to any MessageConverter implementation to convert the message to and from some object that can be serialized and deserialized by the configured serializer and deserializer.

Spring Integration provides a MapMessageConverter, which allows the specification of a list of headers that are 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 and contains the selected headers.

The second step is to provide a serializer and a deserializer that can convert between a Map and some wire format. This can be a custom Serializer or Deserializer, which you typically need if the peer system is not a Spring Integration application.

Spring Integration provides a MapJsonSerializer to convert a Map to and from JSON. It uses a Spring Integration JsonObjectMapper. You can provide a custom JsonObjectMapper if needed. By default, the serializer inserts a linefeed (0x0a) character between objects. See the Javadoc for more information.

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

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

A message sent with the preceding configuration, with a payload of 'something' would appear on the wire as follows:

The pool size attribute is no longer used. Previously, it specified the size of the default thread pool when a task-executor was not specified. It was also used to set the connection backlog on server sockets. The first function is no longer needed (see the next paragraph). The second function is replaced by the backlog attribute.

Previously, when using a fixed thread pool task executor (which was the default) with NIO, it was possible to get a deadlock and processing would stop. The problem occurred when a buffer was full, a thread reading from the socket was trying to add more data to the buffer, and no threads were available to make space in the buffer. This only occurred with a very small pool size, but it could be possible under extreme conditions. Since 2.2, two changes have eliminated this problem. First, the default task executor is a cached thread pool executor. Second, deadlock detection logic has been added such that, if thread starvation occurs, instead of deadlocking, an exception is thrown, thus releasing the deadlocked resources.

You should keep in mind some important considerations when you use 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 do not use a fixed thread pool.

With NIO connections, there are three distinct task types. The I/O selector processing is performed on one dedicated thread (detecting events, accepting new connections, and dispatching the I/O read operations to other threads by using the task executor). When an I/O reader thread (to which the read operation is dispatched) reads data, it hands off to another thread to assemble the incoming message. Large messages can take several reads to complete. These “assembler” threads can block while waiting for data. When a new read event occurs, the reader determines if this socket already has an assembler and, if not, runs a new one. When the assembly process is complete, the assembler thread is returned to the pool.

This can cause a deadlock when the pool is exhausted, 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 by using the executor. The queue is full, so the selector thread itself starts the read process. Now it detects that there is no 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, waiting for the data to be read, which never happens. The connection factory is now deadlocked because the selector thread cannot handle new events.

To avoid this deadlock, we must avoid the selector (or reader) threads performing the assembly task. We want to use separate 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 an AbortPolicy (ABORT when using <task>). When an I/O task cannot be completed, it is deferred for a short time and continually retried until it can be completed and have an assembler allocated. By default, the delay is 100ms, but you can change it by setting the readDelay property on the connection factory (read-delay when configuring with the XML namespace).

The following three examples shows how to configure the composite executor:

Regardless of whether you use NIO, 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 key stores each:

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.

The following example configures an SSL connection:

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

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

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

Starting with version 5.0.8, you can configure whether or not to enable host verification. Starting with version 5.1, it is enabled by default; the mechanism to disable it depends on whether or not you are using NIO.

Host verification is used to ensure the server you are connected to matches information in the certificate, even if the certificate is trusted.

When using NIO, configure the DefaultTcpNioSSLConnectionSupport, for example.

The second constructor argument disables host verification. The connectionSupport bean is then injected into the NIO connection factory.

When not using NIO, the configuration is in the TcpSocketSupport:

Again, the constructor argument disables host verification.

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

To enable client certificate authentication when you use SSL, the technique depends on whether you use NIO. When you do not NIO , provide a custom TcpSocketSupport implementation to post-process the server socket:

(When you use XML configuration, provide a reference to your bean by setting the socket-support attribute).

When you use NIO, provide a custom TcpNioSslConnectionSupport implementation to post-process the SSLEngine, as the following example shows:

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