If you do not set the channel attribute, a DirectChannel is created and registered in the application context with the value of this id attribute as the bean name.
In this case, the endpoint is registered with the bean name id plus .adapter . |
| 2 | Identifies the channel attached to this adapter. |
| 3 | The MessageChannel bean reference to which ErrorMessage instances should be sent. |
| 4 | See the same option on the <int-stomp:outbound-channel-adapter> . |
| 5 | Comma-separated list of names of STOMP Headers to be mapped from the STOMP frame headers.
You can only provide this if the header-mapper reference is not set.
The values in this list can also be simple patterns to be matched against the header names (for example, myheader* or *myheader ).
A special token (STOMP_INBOUND_HEADERS ) represents all the standard STOMP headers (content-length, receipt, heart-beat, and so on).
They are included by default.
If you want to add your own headers and want the standard headers to also be mapped, you must also include this token or provide your own HeaderMapper implementation using header-mapper . |
| 6 | See the same option on the <int-stomp:outbound-channel-adapter> . |
| 7 | Comma-separated list of STOMP destination names to subscribe.
The list of destinations (and therefore subscriptions) can be modified at runtime through the addDestination() and removeDestination() @ManagedOperation annotations. |
| 8 | Maximum amount of time (in milliseconds) to wait when sending a message to the channel if the channel can block.
For example, a QueueChannel can block until space is available if its maximum capacity has been reached. |
| 9 | Fully qualified name of the Java type for the target payload to convert from the incoming STOMP frame.
It defaults to String.class . |
| 10 | See the same option on the <int-stomp:outbound-channel-adapter> . |
| 11 | See the same option on the <int-stomp:outbound-channel-adapter> . |
32. Stream Support
In many cases, application data is obtained from a stream. It is not recommended to send a reference to a stream as a message payload to a consumer. Instead, messages are created from data that is read from an input stream, and message payloads are written to an output stream one by one.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-stream:5.1.5.RELEASE"
32.1. Reading from Streams
Spring Integration provides two adapters for streams.
Both ByteStreamReadingMessageSource
and CharacterStreamReadingMessageSource
implement MessageSource
.
By configuring one of these within a channel-adapter element, the polling period can be configured and the message bus can automatically detect and schedule them.
The byte stream version requires an InputStream
, and the character stream version requires a Reader
as the single constructor argument.
The ByteStreamReadingMessageSource
also accepts the 'bytesPerMessage' property to determine how many bytes it tries to read into each Message
.
The default value is 1024.
The following example creates an input stream that creates messages that each contain 2048 bytes:
<bean class="org.springframework.integration.stream.ByteStreamReadingMessageSource">
<constructor-arg ref="someInputStream"/>
<property name="bytesPerMessage" value="2048"/>
</bean>
<bean class="org.springframework.integration.stream.CharacterStreamReadingMessageSource">
<constructor-arg ref="someReader"/>
</bean>
The CharacterStreamReadingMessageSource
wraps the reader in a BufferedReader
(if it is not one already).
You can set the buffer size used by the buffered reader in the second constructor argument.
Starting with version 5.0, a third constructor argument (blockToDetectEOF
) controls the behavior of the CharacterStreamReadingMessageSource
.
When false
(the default), the receive()
method checks whether the reader is ready()
and returns null if not.
EOF (end of file) is not detected in this case.
When true
, the receive()
method blocks until data is available or EOF is detected on the underlying stream.
When EOF is detected, a StreamClosedEvent
(application event) is published.
You can consume this event with a bean that implements ApplicationListener<StreamClosedEvent>
.
To facilitate EOF detection, the poller thread blocks in the receive() method until either data arrives or EOF is detected.
|
The poller continues to publish an event on each poll once EOF has been detected.
The application listener can stop the adapter to prevent this.
The event is published on the poller thread.
Stopping the adapter causes the thread to be interrupted.
If you intend to perform some interruptible task after stopping the adapter, you must either perform the stop() on a different thread or use a different thread for those downstream activities.
Note that sending to a QueueChannel is interruptible, so, if you wish to send a message from the listener, do it before stopping the adapter.
|
This facilitates “piping” or redirecting data to stdin
, as the following two examples shows:
cat myfile.txt | java -jar my.jar
java -jar my.jar < foo.txt
This approach lets the application terminate when the pipe is closed.
Four convenient factory methods are available:
public static final CharacterStreamReadingMessageSource stdin() { ... }
public static final CharacterStreamReadingMessageSource stdin(String charsetName) { ... }
public static final CharacterStreamReadingMessageSource stdinPipe() { ... }
public static final CharacterStreamReadingMessageSource stdinPipe(String charsetName) { ... }
32.2. Writing to Streams
For target streams, you can use either of two implementations: ByteStreamWritingMessageHandler
or CharacterStreamWritingMessageHandler
.
Each requires a single constructor argument (OutputStream
for byte streams or Writer
for character streams), and each provides a second constructor that adds the optional 'bufferSize'.
Since both of these ultimately implement the MessageHandler
interface, you can reference them from a channel-adapter
configuration, as described in Channel Adapter.
<bean class="org.springframework.integration.stream.ByteStreamWritingMessageHandler">
<constructor-arg ref="someOutputStream"/>
<constructor-arg value="1024"/>
</bean>
<bean class="org.springframework.integration.stream.CharacterStreamWritingMessageHandler">
<constructor-arg ref="someWriter"/>
</bean>
32.3. Stream Namespace Support
Spring Integration defines a namespace to reduce the configuration needed for stream-related channel adapters. The following schema locations are needed to use it:
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/stream
https://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
The following code snippet shows the different configuration options that are supported to configure the inbound channel adapter:
<int-stream:stdin-channel-adapter id="adapterWithDefaultCharset"/>
<int-stream:stdin-channel-adapter id="adapterWithProvidedCharset" charset="UTF-8"/>
Starting with version 5.0, you can set the detect-eof
attribute, which sets the blockToDetectEOF
property.
See Reading from Streams for more information.
To configure the outbound channel adapter, you can use the namespace support as well. The following example shows the different configuration for an outbound channel adapters:
<int-stream:stdout-channel-adapter id="stdoutAdapterWithDefaultCharset"
channel="testChannel"/>
<int-stream:stdout-channel-adapter id="stdoutAdapterWithProvidedCharset" charset="UTF-8"
channel="testChannel"/>
<int-stream:stderr-channel-adapter id="stderrAdapter" channel="testChannel"/>
<int-stream:stdout-channel-adapter id="newlineAdapter" append-newline="true"
channel="testChannel"/>
33. Syslog Support
Spring Integration 2.2 introduced the syslog transformer: SyslogToMapTransformer
.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-syslog</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-syslog:5.1.5.RELEASE"
This transformer, together with a UDP
or TCP
inbound adapter, could be used to receive and analyze syslog records from other hosts.
The transformer creates a message payload that contains a map of the elements from the syslog message.
Spring Integration 3.0 introduced convenient namespace support for configuring a syslog inbound adapter in a single element.
Starting with version 4.1.1, the framework now supports the extended syslog format, as specified in RFC 5424>.
In addition, when using TCP and RFC5424, both octet counting
and non-transparent framing
described in RFC 6587 are supported.
33.1. Syslog Inbound Channel Adapter
This element encompasses a UDP
or TCP
inbound channel adapter and a MessageConverter
to convert the syslog message to a Spring Integration message.
The DefaultMessageConverter
delegates to the SyslogToMapTransformer
, creating a message with its payload being the Map
of syslog fields.
In addition, all fields except the message are also made available as headers in the message and are prefixed with syslog_
.
In this mode, only RFC 3164 (BSD) syslogs are supported.
Since version 4.1, the DefaultMessageConverter
has a property called asMap
(the default is true
).
When it is false
, the converter leaves the message payload as the original complete syslog message (in a byte[]
) while still setting the headers.
Since version 4.1.1, RFC 5424 is also supported, by using the RFC5424MessageConverter
.
In this case, the fields are not copied as headers, unless asMap
is set to false
, in which case the original message is the payload and the decoded fields are headers.
To use RFC 5424 with a TCP transport, you must provide additional configuration to enable the different framing techniques described in RFC 6587.
The adapter needs a TCP connection factory that is configured with a RFC6587SyslogDeserializer .
By default, this deserializer handles octet counting and non-transparent framing by using a linefeed (LF) to delimit syslog messages.
It uses a ByteArrayLfSerializer when octet counting is not detected.
To use different non-transparent framing, you can provide it with some other deserializer.
While the deserializer can support both octet counting and non-transparent framing , only one form of the latter is supported.
If asMap is false on the converter, you must set the retainOriginal constructor argument in the RFC6587SyslogDeserializer .
|
33.1.1. Example Configuration
The following example defines a UDP
adapter that sends messages to the syslogIn
channel (the adapter bean name is syslogIn.adapter
):
<int-syslog:inbound-channel-adapter id="syslogIn" port="1514" />
The adapter listens on port 1514
.
The following example defines a UDP
adapter that sends messages to the fromSyslog
channel (the adapter bean name is syslogIn
):
<int-syslog:inbound-channel-adapter id="syslogIn"
channel="fromSyslog" port="1514" />
The adapter listens on port 1514
.
The following example defines a TCP
adapter that sends messages to channel syslogIn
(the adapter bean name is syslogIn.adapter
):
<int-syslog:inbound-channel-adapter id="bar" protocol="tcp" port="1514" />
The adapter listens on port 1514
.
Note the addition of the protocol
attribute.
This attribute can contain udp
or tcp
.
It defaults to udp
.
The following example shows a UDP
adapter that sends messages to channel fromSyslog
:
<int-syslog:inbound-channel-adapter id="udpSyslog"
channel="fromSyslog"
auto-startup="false"
phase="10000"
converter="converter"
send-timeout="1000"
error-channel="errors">
<int-syslog:udp-attributes port="1514" lookup-host="false" />
</int-syslog:inbound-channel-adapter>
The preceding example also shows two SmartLifecycle
attributes: auto-startup
and phase
.
It has a reference to a custom org.springframework.integration.syslog.MessageConverter
with an ID of converter
and an error-channel
.
Also notice the udp-attributes
child element.
You can set various UDP attributes here, as defined in UDP Inbound Channel Adapter Attributes.
When you use the udp-attributes element, you must provide the port attribute there rather than on the inbound-channel-adapter element itself.
|
The following example shows a TCP
adapter that sends messages to channel fromSyslog
:
<int-syslog:inbound-channel-adapter id="TcpSyslog"
protocol="tcp"
channel="fromSyslog"
connection-factory="cf" />
<int-ip:tcp-connection-factory id="cf" type="server" port="1514" />
It also shows how to reference an externally defined connection factory, which can be used for advanced configuration (socket keep-alive and other uses). For more information, see TCP Connection Factories.
The externally configured connection-factory must be of type server , and the port is defined there rather than on the inbound-channel-adapter element itself.
|
The following example shows a TCP
adapter that sends messages to channel fromSyslog
:
<int-syslog:inbound-channel-adapter id="rfc5424Tcp"
protocol="tcp"
channel="fromSyslog"
connection-factory="cf"
converter="rfc5424" />
<int-ip:tcp-connection-factory id="cf"
using-nio="true"
type="server"
port="1514"
deserializer="rfc6587" />
<bean id="rfc5424" class="org.springframework.integration.syslog.RFC5424MessageConverter" />
<bean id="rfc6587" class="org.springframework.integration.syslog.inbound.RFC6587SyslogDeserializer" />
The preceding example is configured to use the RFC 5424
converter and is configured with a reference to an externally defined connection factory with the RFC 6587
deserializer (required for RFC 5424).
34. TCP and UDP Support
Spring Integration provides channel adapters for receiving and sending messages over internet protocols. Both UDP (User Datagram Protocol) and TCP (Transmission Control Protocol) adapters are provided. Each adapter provides for one-way communication over the underlying protocol. In addition, Spring Integration provides simple inbound and outbound TCP gateways. These are used when two-way communication is needed.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ip</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-ip:5.1.5.RELEASE"
34.1. Introduction
Two flavors each of UDP inbound and outbound channel adapters are provided:
-
UnicastSendingMessageHandler
sends a datagram packet to a single destination. -
UnicastReceivingChannelAdapter
receives incoming datagram packets. -
MulticastSendingMessageHandler
sends (broadcasts) datagram packets to a multicast address. -
MulticastReceivingChannelAdapter
receives incoming datagram packets by joining to a multicast address.
TCP inbound and outbound channel adapters are provided:
-
TcpSendingMessageHandler
sends messages over TCP. -
TcpReceivingChannelAdapter
receives messages over TCP.
An inbound TCP gateway is provided. It allows for simple request-response processing. While the gateway can support any number of connections, each connection can only be processed serially. The thread that reads from the socket waits for, and sends, the response before reading again. If the connection factory is configured for single use connections, the connection is closed after the socket times out.
An outbound TCP gateway is provided. It allows for simple request-response processing. If the associated connection factory is configured for single-use connections, a new connection is immediately created for each new request. Otherwise, if the connection is in use, the calling thread blocks on the connection until either a response is received or a timeout or I/O error occurs.
The TCP and UDP inbound channel adapters and the TCP inbound gateway support the error-channel
attribute.
This provides the same basic functionality as described in Enter the GatewayProxyFactoryBean
.
34.2. UDP Adapters
This section describes how to configure and use the UDP adapters.
34.2.1. Outbound UDP Adapters (XML Configuration)
The following example configures a UDP outbound channel adapter:
<int-ip:udp-outbound-channel-adapter id="udpOut"
host="somehost"
port="11111"
multicast="false"
channel="exampleChannel"/>
When setting multicast to true , you should also provide the multicast address in the host attribute.
|
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
.
The preceding example shows an outbound channel adapter that adds length checking to the datagram packets:
<int-ip:udp-outbound-channel-adapter id="udpOut"
host="somehost"
port="11111"
multicast="false"
check-length="true"
channel="exampleChannel"/>
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.
The following example shows an outbound channel adapter that adds length checking to the datagram packets and waits for an acknowledgment:
<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"/>
Setting acknowledge to true implies that the recipient of the packet can interpret the header added to the packet containing acknowledgment data (host and port).
Most likely, the recipient is a Spring Integration inbound channel adapter.
|
When multicast is true, an additional attribute (min-acks-for-success ) specifies how many acknowledgments must be received within the ack-timeout .
|
Starting with version 4.3, you can set the ackPort
to 0
, in which case the operating system chooses the port.
34.2.2. Outbound UDP Adapters (Java Configuration)
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);
}
(or MulticastSendingChannelAdapter
for multicast).
34.2.3. Outbound UDP Adapters (Java DSL Configuration)
The following example shows how to configure an outbound UDP adapter with the Java DSL:
@Bean
public IntegrationFlow udpOutFlow() {
return IntegrationFlows.from("udpOut")
.handle(Udp.outboundAdapter("localhost", 1234))
.get();
}
34.2.4. Inbound UDP Adapters (XML Configuration)
The following example shows how to configure a basic unicast inbound udp channel adapter.
<int-ip:udp-inbound-channel-adapter id="udpReceiver"
channel="udpOutChannel"
port="11111"
receive-buffer-size="500"
multicast="false"
check-length="true"/>
The following example shows how to configure a basic multicast inbound udp channel adapter:
<int-ip:udp-inbound-channel-adapter id="udpReceiver"
channel="udpOutChannel"
port="11111"
receive-buffer-size="500"
multicast="true"
multicast-address="<IP_ADDRESS>"
check-length="true"/>
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.
You can override this default behavior by setting the lookup-host
attribute to false
.
34.2.5. Inbound UDP Adapters (Java Configuration)
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;
}
The following example shows how to configure an inbound UDP adapter with the Java DSL:
34.2.6. Inbound UDP Adapters (Java DSL Configuration)
@Bean
public IntegrationFlow udpIn() {
return IntegrationFlows.from(Udp.inboundAdapter(11111))
.channel("udpChannel")
.get();
}
34.2.7. Server Listening Events
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.
34.2.8. Advanced Outbound Configuration
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:
<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"
<EMAIL_ADDRESS>destination-expression="headers['ip_packetAddress']"
channel="out" />
The following example shows the equivalent configuration with the Java DSL:
@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();
}
34.3. TCP Connection Factories
34.3.1. Overview
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.
Reply messages are routed to the connection only if the reply contains the ip_connectionId header that was inserted into the original message by the connection factory.
|
| This is the extent of message correlation performed when sharing connection factories between inbound and outbound adapters. Such sharing allows for asynchronous two-way communication over TCP. By default, only payload information is transferred using TCP. Therefore, any message correlation must be performed by downstream components such as aggregators or other endpoints. Support for transferring selected headers was introduced in version 3.0. For more information, see TCP Message Correlation. |
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:
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"/>
The following example shows a simple server connection factory that uses java.nio.channel.SocketChannel
connections:
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"
using-nio="true"/>
Starting with Spring Integration version 4.2, if the server is configured to listen on a random port (by setting the port to 0 ), you can get the actual port chosen by the OS by using getPort() .
Also, getServerSocketAddress() lets you get the complete SocketAddress .
See the Javadoc for the TcpServerConnectionFactory interface for more information.
|
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="1234"
single-use="true"
so-timeout="10000"/>
The following example shows a client connection factory that uses java.net.Socket
connections and creates a new connection for each message:
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="1234"
single-use="true"
so-timeout="10000"
using-nio=true/>
34.3.2. Message Demarcation (Serializers and Deserializers)
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.
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.
Before version 4.2.2, when using non-blocking I/O (NIO), this serializer treated a timeout (during read) as an end of file, and the data read so far was emitted as a message.
This is unreliable and should not be used to delimit messages.
It now treats such conditions as an exception.
In the unlikely event that you use it this way, you can restore the previous behavior by setting the treatTimeoutAsEndOfMessage constructor argument to true .
|
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.
| This deserializer should only be used when the peer is trusted; it is susceptible to a DoS attach due to out of memory conditions. |
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 Jackson ObjectMapper cannot demarcate messages in the stream.
Therefore, the MapJsonSerializer needs to delegate to another serializer or deserializer to handle message demarcation.
By default, a ByteArrayLfSerializer is used, resulting in messages with a format of <json><LF> on the wire, but you can configure it to use others instead.
(The next example shows how to do so.)
|
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:
<bean id="javaSerializer"
class="org.springframework.core.serializer.DefaultSerializer" />
<bean id="javaDeserializer"
class="org.springframework.core.serializer.DefaultDeserializer" />
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"
deserializer="javaDeserializer"
serializer="javaSerializer"/>
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 hostnames 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
.
| You can also modify the attributes of sockets and socket factories. See SSL/TLS Support. As noted there, such modifications are possible whether or not SSL is being used. |
34.3.3. Custom Serializers and Deserializers
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.
34.3.4. TCP Caching Client Connection Factory
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.
34.3.5. TCP Failover Client Connection Factory
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:
<bean id="failCF" class="o.s.i.ip.tcp.connection.FailoverClientConnectionFactory">
<constructor-arg>
<list>
<ref bean="clientFactory1"/>
<ref bean="clientFactory2"/>
</list>
</constructor-arg>
</bean>
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.
|
34.3.6. TCP Thread Affinity Connection Factory
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:
@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;
}
34.4. TCP Connection Interceptors
You can configure connection factories with a reference to a TcpConnectionInterceptorFactoryChain
.
You can use interceptors to add behavior to connections, such as negotiation, security, and other options.
No interceptors are currently provided by the framework, but see InterceptedSharedConnectionTests
in the source repository for an example.
The HelloWorldInterceptor
used in the test case works as follows:
The interceptor is first configured with a client connection factory. When the first message is sent over an intercepted connection, the interceptor sends 'Hello' over the connection and expects to receive 'world!'. When that occurs, the negotiation is complete and the original message is sent. Further messages that use the same connection are sent without any additional negotiation.
When configured with a server connection factory, the interceptor requires the first message to be 'Hello' and, if it is, returns 'world!'. Otherwise it throws an exception that causes the connection to be closed.
All TcpConnection
methods are intercepted.
Interceptor instances are created for each connection by an interceptor factory.
If an interceptor is stateful, the factory should create a new instance for each connection.
If there is no state, the same interceptor can wrap each connection.
Interceptor factories are added to the configuration of an interceptor factory chain, which you can provide to a connection factory by setting the interceptor-factory
attribute.
Interceptors must extend TcpConnectionInterceptorSupport
.
Factories must implement the TcpConnectionInterceptorFactory
interface.
TcpConnectionInterceptorSupport
has passthrough methods.
By extending this class, you only need to implement those methods you wish to intercept.
The following example shows how to configure a connection interceptor factory chain:
<bean id="helloWorldInterceptorFactory"
class="o.s.i.ip.tcp.connection.TcpConnectionInterceptorFactoryChain">
<property name="interceptors">
<array>
<bean class="o.s.i.ip.tcp.connection.HelloWorldInterceptorFactory"/>
</array>
</property>
</bean>
<int-ip:tcp-connection-factory id="server"
type="server"
port="12345"
using-nio="true"
single-use="true"
interceptor-factory-chain="helloWorldInterceptorFactory"/>
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="12345"
single-use="true"
so-timeout="100000"
using-nio="true"
interceptor-factory-chain="helloWorldInterceptorFactory"/>
34.5. TCP Connection Events
Beginning with version 3.0, changes to TcpConnection
instances are reported by TcpConnectionEvent
instances.
TcpConnectionEvent
is a subclass of ApplicationEvent
and can thus be received by any ApplicationListener
defined in the ApplicationContext
— for example an event inbound channel adapter.
TcpConnectionEvents
have the following properties:
-
connectionId
: The connection identifier, which you can use in a message header to send data to the connection. -
connectionFactoryName
: The bean name of the connection factory to which the connection belongs. -
throwable
: TheThrowable
(forTcpConnectionExceptionEvent
events only). -
source
: TheTcpConnection
. You can use this, for example, to determine the remote IP Address withgetHostAddress()
(cast required).
In addition, since version 4.0, the standard deserializers discussed in TCP Connection Factories now emit TcpDeserializationExceptionEvent
instances when they encounter problems while decoding the data stream.
These events contain the exception, the buffer that was in the process of being built, and an offset into the buffer (if available) at the point where the exception occurred.
Applications can use a normal ApplicationListener
or an ApplicationEventListeningMessageProducer
(see Receiving Spring Application Events) to capture these events, allowing analysis of the problem.
Starting with versions 4.0.7 and 4.1.3, TcpConnectionServerExceptionEvent
instances are published whenever an unexpected exception occurs on a server socket (such as a BindException
when the server socket is in use).
These events have a reference to the connection factory and the cause.
Starting with version 4.2, TcpConnectionFailedCorrelationEvent
instances are published whenever an endpoint (inbound gateway or
collaborating outbound channel adapter) receives a message that cannot be routed to a connection because the
ip_connectionId
header is invalid.
Outbound gateways also publish this event when a late reply is received (the sender thread has timed out).
The event contains the connection ID as well as an exception in the cause
property, which contains the failed message.
Starting with version 4.3, a TcpConnectionServerListeningEvent
is emitted when a server connection factory is started.
This is useful when the factory 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 connects to the socket.
| To avoid delaying the listening thread from accepting connections, the event is published on a separate thread. |
Starting with version 4.3.2, a TcpConnectionFailedEvent
is emitted whenever a client connection cannot be created.
The source of the event is the connection factory, which you can use to determine the host and port to which the connection could not be established.
34.6. TCP Adapters
TCP inbound and outbound channel adapters that use connection factories mentioned earlier are provided.
These adapters have two relevant attributes: connection-factory
and channel
.
The connection-factory
attribute indicates which connection factory is to be used to manage connections for the adapter.
The channel
attribute specifies the channel on which messages arrive at an outbound adapter and on which messages are placed by an inbound adapter.
While both inbound and outbound adapters can share a connection factory, server connection factories are always “owned” by an inbound adapter.
Client connection factories are always “owned” by an outbound adapter.
Only one adapter of each type may get a reference to a connection factory.
The following example shows how to define client and server TCP connection factories:
<bean id="javaSerializer"
class="org.springframework.core.serializer.DefaultSerializer"/>
<bean id="javaDeserializer"
class="org.springframework.core.serializer.DefaultDeserializer"/>
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"
deserializer="javaDeserializer"
serializer="javaSerializer"
using-nio="true"
single-use="true"/>
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="#{server.port}"
single-use="true"
so-timeout="10000"
deserializer="javaDeserializer"
serializer="javaSerializer"/>
<int:channel id="input" />
<int:channel id="replies">
<int:queue/>
</int:channel>
<int-ip:tcp-outbound-channel-adapter id="outboundClient"
channel="input"
connection-factory="client"/>
<int-ip:tcp-inbound-channel-adapter id="inboundClient"
channel="replies"
connection-factory="client"/>
<int-ip:tcp-inbound-channel-adapter id="inboundServer"
channel="loop"
connection-factory="server"/>
<int-ip:tcp-outbound-channel-adapter id="outboundServer"
channel="loop"
connection-factory="server"/>
<int:channel id="loop"/>
In the preceding configuration, messages arriving in the input
channel are serialized over connections created by client
connection factory, received at the server, and placed on the loop
channel.
Since loop
is the input channel for outboundServer
, the message is looped back over the same connection, received by inboundClient
, and deposited in the replies
channel.
Java serialization is used on the wire.
Normally, inbound adapters use a type="server"
connection factory, which listens for incoming connection requests.
In some cases, you may want to establish the connection in reverse, such that the inbound adapter connects to an external server and then waits for inbound messages on that connection.
This topology is supported by setting client-mode="true"
on the inbound adapter.
In this case, the connection factory must be of type client
and must have single-use
set to false
.
Two additional attributes support this mechanism.
The retry-interval
specifies (in milliseconds) how often the framework attempts to reconnect after a connection failure.
scheduler
supplies a TaskScheduler
to schedule the connection attempts and to test that the connection is still active.
If you don’t provide a scheduler, the framework’s default taskScheduler bean is used.
For an outbound adapter, the connection is normally established when the first message is sent.
client-mode="true"
on an outbound adapter causes the connection to be established when the adapter is started.
By default, adapters are automatically started.
Again, the connection factory must be of type client
and have single-use="false"
.
retry-interval
and scheduler
are also supported.
If a connection fails, it is re-established either by the scheduler or when the next message is sent.
For both inbound and outbound, if the adapter is started, you can force the adapter to establish a connection by sending a <control-bus />
command: @adapter_id.retryConnection()
.
Then you can examine the current state with @adapter_id.isClientModeConnected()
.
34.7. TCP Gateways
The inbound TCP gateway TcpInboundGateway
and outbound TCP gateway TcpOutboundGateway
use a server and client connection factory, respectively.
Each connection can process a single request or response at a time.
The inbound gateway, after constructing a message with the incoming payload and sending it to the requestChannel
, waits for a response and sends the payload from the response message by writing it to the connection.
For the inbound gateway, you must retain or populate, the ip_connectionId header, because it is used to correlate the message to a connection.
Messages that originate at the gateway automatically have the header set.
If the reply is constructed as a new message, you need to set the header.
The header value can be captured from the incoming message.
|
As with inbound adapters, inbound gateways normally use a type="server"
connection factory, which listens for incoming connection requests.
In some cases, you may want to establish the connection in reverse, such that the inbound gateway connects to an external server and then waits for and replies to inbound messages on that connection.
This topology is supported by using client-mode="true"
on the inbound gateway.
In this case, the connection factory must be of type client
and must have single-use
set to false
.
Two additional attributes support this mechanism.
retry-interval
specifies (in milliseconds) how often the framework tries to reconnect after a connection failure.
scheduler
supplies a TaskScheduler
to schedule the connection attempts and to test that the connection is still active.
If the gateway is started, you may force the gateway to establish a connection by sending a <control-bus /> command: @adapter_id.retryConnection()
and examine the current state with @adapter_id.isClientModeConnected()
.
The outbound gateway, after sending a message over the connection, waits for a response, constructs a response message, and puts it on the reply channel. Communications over the connections are single-threaded. Only one message can be handled at a time. If another thread attempts to send a message before the current response has been received, it blocks until any previous requests are complete (or time out). If, however, the client connection factory is configured for single-use connections, each new request gets its own connection and is processed immediately. The following example configures an inbound TCP gateway:
<int-ip:tcp-inbound-gateway id="inGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfServer"
reply-timeout="10000"/>
If a connection factory configured with the default serializer or deserializer is used, messages is \r\n
delimited data and the gateway can be used by a simple client such as telnet.
The following example shows an outbound TCP gateway:
<int-ip:tcp-outbound-gateway id="outGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfClient"
request-timeout="10000"
remote-timeout="10000"/> <!-- or e.g.
remote-timeout-expression="headers['timeout']" -->
client-mode
is not currently available with the outbound gateway.
34.8. TCP Message Correlation
One goal of the IP endpoints is to provide communication with systems other than Spring Integration applications. For this reason, only message payloads are sent and received by default. Since 3.0, you can transfer headers by using JSON, Java serialization, or custom serializers and deserializers. See 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. Later in this document, 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).
34.8.1. Gateways
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.
34.8.2. Collaborating Outbound and Inbound Channel Adapters
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 server side, you must populate the ip_connectionId header, because it is used to correlate the message to a connection.
Messages that originate at the inbound adapter automatically have the header set.
If you wish to construct other messages to send, you need to set the header.
You can get the header value from an incoming 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[]
.
|
Before the 2.2 release, when collaborating channel adapters used a client connection factory, the 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 |
34.8.3. Transferring Headers
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.
The JsonObjectMapper uses whichever version of Jackson is on the classpath.
|
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:
<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 preceding configuration, with a payload of 'something' would appear on the wire as follows:
{"headers":{"correlationId":"things","sequenceSize":5,"sequenceNumber":1},"payload":"something"}
34.9. About Non-blocking I/O (NIO)
Using NIO (see using-nio
in IP Configuration Attributes) avoids dedicating a thread to read from each socket.
For a small number of sockets, you are likely to find that not using NIO, together with an asynchronous handoff (such as to a QueueChannel
), performs as well as or better than using NIO.
You should consider using NIO when handling a large number of connections. However, the use of NIO has some other ramifications. A pool of threads (in the task executor) is shared across all the sockets. Each incoming message is assembled and sent to the configured channel as a separate unit of work on a thread selected from that pool. Two sequential messages arriving on the same socket might be processed by different threads. This means that the order in which the messages are sent to the channel is indeterminate. Strict ordering of the messages arriving on the socket is not maintained.
For some applications, this is not an issue.
For others, it is a problem.
If you require strict ordering, consider setting using-nio
to false
and using an asynchronous handoff.
Alternatively, you can insert a resequencer downstream of the inbound endpoint to return the messages to their proper sequence.
If you set apply-sequence
to true
on the connection factory, messages arriving on a TCP connection have sequenceNumber
and correlationId
headers set.
The resequencer uses these headers to return the messages to their proper sequence.
Starting with version 5.1.4, priority is given to accepting new connections over reading from existing connections.
This should, generally, have little impact unless you have a very high rate of new incoming connections.
If you wish to revert to the previous behavior of giving reads priority, set the multiAccept property on the TcpNioServerConnectionFactory to false .
|
34.9.1. Pool Size
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.
| Now that the default task executor is unbounded, it is possible that an out-of-memory condition might occur with high rates of incoming messages, if message processing takes extended time. If your application exhibits this type of behavior, you should use a pooled task executor with an appropriate pool size, but see the next section. |
34.9.2. Thread Pool Task Executor with CALLER_RUNS
Policy
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:
@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>
34.10. SSL/TLS Support
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.
The default implementations of these interfaces provide for the simplest way to get started with secure communications.
34.10.1. Getting Started
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 keystores each:
-
A keystore that contains private and public key pairs to identify the peer
-
A truststore that contains the public keys for peers that are trusted. See the documentation for the
keytool
utility provided with the JDK. The essential steps are-
Create a new key pair and store it in a keystore.
-
Export the public key.
-
Import the public key into the peer’s truststore.
-
Repeat for the other peer.
-
| 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.
The following example configures an SSL connection:
<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
(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.
34.10.2. Host Verification
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.
@Bean
public DefaultTcpNioSSLConnectionSupport connectionSupport() {
DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
"test.truststore.ks", "secret", "secret");
sslContextSupport.setProtocol("SSL");
DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport =
new DefaultTcpNioSSLConnectionSupport(sslContextSupport, false);
return tcpNioConnectionSupport;
}
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
:
connectionFactory.setTcpSocketSupport(new DefaultTcpSocketSupport(false));
Again, the constructor argument disables host verification.
34.11. Advanced Techniques
This section covers advanced techniques that you may find to be helpful in certain situations.
34.11.1. Strategy Interfaces
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:
-
TcpSSLContextSupport
-
TcpSocketFactorySupport
-
TcpSocketSupport
-
TcpNetConnectionSupport
-
TcpNioConnectionSupport
The TcpSSLContextSupport
Strategy Interface
The following listing shows the TcpSSLContextSupport
strategy interface:
public interface TcpSSLContextSupport {
SSLContext getSSLContext() throws Exception;
}
Implementations of the TcpSSLContextSupport
interface are responsible for creating an SSLContext
object.
The implementation provided by the framework is the DefaultTcpSSLContextSupport
, described earlier.
If you require different behavior, implement this interface and provide the connection factory with a reference to a bean of your class' implementation.
The TcpSocketFactorySupport
Strategy Interface
The following listing shows the definition of the TcpSocketFactorySupport
strategy interface:
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 uses the JDK’s default factories.
The second implementation is DefaultTcpNetSSLSocketFactorySupport
.
By default, this is used when an ssl-context-support
attribute is defined.
It uses the SSLContext
created by that bean to create the socket factories.
This interface applies only if using-nio is false .
NIO does not use socket factories.
|
The TcpSocketSupport
Strategy Interface
The following listing shows the definition of the TcpSocketSupport
strategy interface:
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 you use NIO or not.
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 by setting the socket-factory-support
and socket-support
attributes, respectively.
The TcpNetConnectionSupport
Strategy Interface
The following listing shows the definition of the TcpNetConnectionSupport
strategy interface:
public interface TcpNetConnectionSupport {
TcpNetConnection createNewConnection(Socket socket,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
}
This interface is invoked to create objects of type TcpNetConnection
(or its subclasses).
The framework provides a single implementation (DefatulTcpNetConnectionSupport
), which, by default, creates simple TcpNetConnection
objects.
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 lets deserializers “unread” (push back) bytes into the stream.
The following trivial example shows how it might be used in a delegating deserializer that “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);
}
}
}
The TcpNioConnectionSupport
Strategy Interface
The following listing shows the definition of the TcpNioConnectionSupport
strategy interface:
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).
Spring Integration provides two implementations: DefaultTcpNioSSLConnectionSupport
and DefaultTcpNioConnectionSupport
.
Which one is used depends on whether SSL is in use.
A common use case is to subclass DefaultTcpNioSSLConnectionSupport
and override postProcessSSLEngine
.
See the SSL client authentication example.
As with the DefaultTcpNetConnectionSupport
, these implementations also support push back.
34.11.2. Example: Enabling SSL Client Authentication
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:
serverFactory.setTcpSocketSupport(new DefaultTcpSocketSupport() {
@Override
public void postProcessServerSocket(ServerSocket serverSocket) {
((SSLServerSocket) serverSocket).setNeedClientAuth(true);
}
});
(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:
@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 you use XML configuration, since version 4.3.7, provide a reference to your bean by setting the nio-connection-support
attribute).
34.12. IP Configuration Attributes
The following table describes attributes that you can set to configure IP connections:
| Attribute Name | Client? | Server? | Allowed Values | Attribute Description |
|---|---|---|---|---|
|
Y |
Y |
client, server |
Determines whether the connection factory is a client or a server. |
|
Y |
N |
The host name or IP address of the destination. |
|
|
Y |
Y |
The port. |
|
|
Y |
Y |
An implementation of |
|
|
Y |
Y |
An implementation of |
|
|
Y |
Y |
|
Whether or not connection uses NIO.
Refer to the |
|
Y |
N |
|
When using NIO, whether or not the connection uses direct buffers.
Refer to the |
|
Y |
Y |
|
When you use NIO, it may be necessary to resequence messages.
When this attribute is set to |
|
Y |
Y |
Defaults to |
|
|
Y |
Y |
See |
|
|
Y |
Y |
See |
|
|
Y |
Y |
|
See |
|
Y |
Y |
Sets |
|
|
Y |
Y |
|
See |
|
Y |
Y |
See |
|
|
N |
Y |
On a multi-homed system, specifies an IP address for the interface to which the socket is bound. |
|
|
Y |
Y |
Specifies a specific executor to be used for socket handling.
If not supplied, an internal cached thread executor is used.
Needed on some platforms that require the use of specific task executors, such as a |
|
|
Y |
Y |
|
Specifies whether a connection can be used for multiple messages.
If |
|
N |
N |
This attribute is no longer used.
For backward compatibility, it sets the backlog, but you should use |
|
|
N |
Y |
Sets the connection backlog for server factories. |
|
|
Y |
Y |
|
Specifies whether reverse lookups are done on IP addresses to convert to host names for use in message headers.
If false, the IP address is used instead.
Default: |
|
Y |
Y |
||
|
Y |
Y |
See |
|
|
Y |
Y |
See |
|
|
Y |
Y |
See SSL/TLS Support. |
|
|
Y |
Y |
See Advanced Techniques. |
|
|
Y |
Y |
long > 0 |
The delay (in milliseconds) before retrying a read after the previous attempt failed due to insufficient threads.
Default: 100.
Only applies if |
The following table describes attributes that you can set to configure UDP inbound channel adapters:
| Attribute Name | Allowed Values | Attribute Description |
|---|---|---|
|
The port on which the adapter listens. |
|
|
|
Whether or not the UDP adapter uses multicast. |
|
When multicast is true, the multicast address to which the adapter joins. |
|
|
Specifies how many packets can be handled concurrently. It only applies if task-executor is not configured. Default: 5. |
|
task-executor |
Specifies a specific executor to be used for socket handling.
If not supplied, an internal pooled executor is used.
Needed on some platforms that require the use of specific task executors such as a |
|
|
The size of the buffer used to receive |
|
|
|
Whether or not a UDP adapter expects a data length field in the packet received. Used to detect packet truncation. |
|
See the |
|
|
Used for UDP acknowledgment packets.
See the setSendBufferSize() methods in |
|
|
See |
|
|
On a multi-homed system, specifies an IP address for the interface to which the socket is bound. |
|
|
If a downstream component throws an exception, the |
|
|
|
Specifies whether reverse lookups are done on IP addresses to convert to host names for use in message headers.
If |
The following table describes attributes that you can set to configure UDP outbound channel adapters:
| Attribute Name | Allowed Values | Attribute Description |
|---|---|---|
|
The host name or ip address of the destination. For multicast udp adapters, the multicast address. |
|
|
The port on the destination. |
|
|
|
Whether or not the udp adapter uses multicast. |
|
|
Whether or not a UDP adapter requires an acknowledgment from the destination.
When enabled, it requires setting the following four attributes: |
|
When |
|
|
When |
|
|
When |
|
|
Defaults to 1. For multicast adapters, you can set this to a larger value, which requires acknowledgments from multiple destinations. |
|
|
|
Whether or not a UDP adapter includes a data length field in the packet sent to the destination. |
|
For multicast adapters, specifies the time-to-live attribute for the |
|
|
See |
|
|
See the |
|
|
Used for UDP acknowledgment packets.
See the |
|
local-address |
On a multi-homed system, for the UDP adapter, specifies an IP address for the interface to which the socket is bound for reply messages. For a multicast adapter, it also determines which interface the multicast packets are sent over. |
|
|
Specifies a specific executor to be used for acknowledgment handling.
If not supplied, an internal single threaded executor is used.
Needed on some platforms that require the use of specific task executors, such as a |
|
|
SpEL expression |
A SpEL expression to be evaluated to determine which |
|
SpEL expression |
A SpEL expression to be evaluated to determine which datagram socket use for sending outgoing UDP packets. |
The following table describes attributes that you can set to configure TCP inbound channel adapters:
| Attribute Name | Allowed Values | Attribute Description |
|---|---|---|
|
The channel to which inbound messages is sent. |
|
|
If the connection factory has a type of |
|
|
If an exception is thrown by a downstream component, the |
|
|
|
When |
|
When in |
|
|
|
Specifies a |
The following table describes attributes that you can set to configure TCP outbound channel adapters:
| Attribute Name | Allowed Values | Attribute Description |
|---|---|---|
|
The channel on which outbound messages arrive. |
|
|
If the connection factory has a type of |
|
|
|
When |
|
When in |
|
|
|
Specifies a |
The following table describes attributes that you can set to configure TCP inbound gateways:
| Attribute Name | Allowed Values | Attribute Description |
|---|---|---|
|
The connection factory must be of type server. |
|
|
The channel to which incoming messages are sent. |
|
|
The channel on which reply messages may arrive. Usually, replies arrive on a temporary reply channel added to the inbound message header. |
|
|
The time in milliseconds for which the gateway waits for a reply. Default: 1000 (1 second). |
|
|
If an exception is thrown by a downstream component, the |
|
|
|
When |
|
When in |
|
|
|
Specifies a |
The following table describes attributes that you can set to configure TCP outbound gateways:
| Attribute Name | Allowed Values | Attribute Description |
|---|---|---|
|
The connection factory must be of type |
|
|
The channel on which outgoing messages arrive. |
|
|
Optional. The channel to which reply messages are sent. |
|
|
The time in milliseconds for which the gateway waits for a reply from the remote system.
Mutually exclusive with |
|
|
A SpEL expression that is evaluated against the message to determine the time in milliseconds for which the gateway waits for a reply from the remote system.
Mutually exclusive with |
|
|
If a single-use connection factory is not being used, the time in milliseconds for which the gateway waits to get access to the shared connection. |
|
|
The time in milliseconds for which the gateway waits when sending the reply to the reply-channel. Only applies if the reply-channel might block (such as a bounded QueueChannel that is currently full). |
34.13. IP Message Headers
This module uses the following MessageHeader
instances:
| Header Name | IpHeaders Constant | Description |
|---|---|---|
|
|
The host name from which a TCP message or UDP packet was received.
If |
|
|
The IP address from which a TCP message or UDP packet was received. |
|
|
The remote port for a UDP packet. |
ip_localInetAddress |
|
The local |
|
|
The remote IP address to which UDP application-level acknowledgments are sent. The framework includes acknowledgment information in the data packet. |
|
|
A correlation ID for UDP application-level acknowledgments. The framework includes acknowledgment information in the data packet. |
|
|
The remote port for a TCP connection. |
|
|
A unique identifier for a TCP connection. Set by the framework for inbound messages. When sending to a server-side inbound channel adapter or replying to an inbound gateway, this header is required so that the endpoint can determine the connection to which to send the message. |
|
|
For information only. When using a cached or failover client connection factory, it contains the actual underlying connection ID. |
|
|
An optional content type for inbound messages
Described after this table.
Note that, unlike the other header constants, this constant is in the |
For inbound messages, ip_hostname
, ip_address
, ip_tcp_remotePort
, and ip_connectionId
are mapped by the default
TcpHeaderMapper
.
If you set the mapper’s addContentTypeHeader
property to true
, the mapper sets the contentType
header (application/octet-stream;charset="UTF-8"
, by default).
You can change the default by setting the contentType
property.
You can add additional headers by subclassing TcpHeaderMapper
and overriding the supplyCustomHeaders
method.
For example, when you use SSL, you can add properties of the SSLSession
by obtaining the session object from the
TcpConnection
object, which is provided as an argument to the supplyCustomHeaders
method.
For outbound messages, String
payloads are converted to byte[]
with the default (UTF-8
) charset.
Set the charset
property to change the default.
When customizing the mapper properties or subclassing, declare the mapper as a bean and provide an instance to the connection factory by using the mapper
property
34.14. Annotation-Based Configuration
The following example from the samples repository shows some of the configuration options available when you use annotations instead of XML:
@EnableIntegration (1)
@IntegrationComponentScan (2)
@Configuration
public static class Config {
@Value(${some.port})
private int port;
@MessagingGateway(defaultRequestChannel="toTcp") (3)
public interface Gateway {
String viaTcp(String in);
}
@Bean
@ServiceActivator(inputChannel="toTcp") (4)
public MessageHandler tcpOutGate(AbstractClientConnectionFactory connectionFactory) {
TcpOutboundGateway gate = new TcpOutboundGateway();
gate.setConnectionFactory(connectionFactory);
gate.setOutputChannelName("resultToString");
return gate;
}
@Bean (5)
public TcpInboundGateway tcpInGate(AbstractServerConnectionFactory connectionFactory) {
TcpInboundGateway inGate = new TcpInboundGateway();
inGate.setConnectionFactory(connectionFactory);
inGate.setRequestChannel(fromTcp());
return inGate;
}
@Bean
public MessageChannel fromTcp() {
return new DirectChannel();
}
@MessageEndpoint
public static class Echo { (6)
@Transformer(inputChannel="fromTcp", outputChannel="toEcho")
public String convert(byte[] bytes) {
return new String(bytes);
}
@ServiceActivator(inputChannel="toEcho")
public String upCase(String in) {
return in.toUpperCase();
}
@Transformer(inputChannel="resultToString")
public String convertResult(byte[] bytes) {
return new String(bytes);
}
}
@Bean
public AbstractClientConnectionFactory clientCF() { (7)
return new TcpNetClientConnectionFactory("localhost", this.port);
}
@Bean
public AbstractServerConnectionFactory serverCF() { (8)
return new TcpNetServerConnectionFactory(this.port);
}
}
| 1 | Standard Spring Integration annotation enabling the infrastructure for an integration application. |
| 2 | Searches for @MessagingGateway interfaces. |
| 3 | The entry point to the client-side of the flow.
The calling application can use @Autowired for this Gateway bean and invoke its method. |
| 4 | Outbound endpoints consist of a MessageHandler and a consumer that wraps it.
In this scenario, the @ServiceActivator configures the endpoint, according to the channel type. |
| 5 | Inbound endpoints (in the TCP/UDP module) are all message-driven and so only need to be declared as simple @Bean instances. |
| 6 | This class provides a number of POJO methods for use in this sample flow (a @Transformer and @ServiceActivator on the server side and a @Transformer on the client side). |
| 7 | The client-side connection factory. |
| 8 | The server-side connection factory. |
35. WebFlux Support
The WebFlux Spring Integration module (spring-integration-webflux
) allows for the execution of HTTP requests and the processing of inbound HTTP requests in a reactive manner.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-webflux</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-webflux:5.1.5.RELEASE"
The io.projectreactor.netty:reactor-netty
dependency must be included in case of non-Servlet-based server configuration.
The WebFlux support consists of the following gateway implementations: WebFluxInboundEndpoint
and WebFluxRequestExecutingMessageHandler
.
The support is fully based on the Spring WebFlux and Project Reactor foundations.
See HTTP Support for more information, since many options are shared between reactive and regular HTTP components.
35.1. WebFlux Inbound Components
Starting with version 5.0, the WebFluxInboundEndpoint
implementation of WebHandler
is provided.
This component is similar to the MVC-based HttpRequestHandlingEndpointSupport
, with which it shares some common options through the newly extracted BaseHttpInboundEndpoint
.
It is used in the Spring WebFlux reactive environment (instead of MVC).
The following example shows a simple implementation of a WebFlux endpoint:
@Configuration
@EnableWebFlux
@EnableIntegration
public class ReactiveHttpConfiguration {
@Bean
public WebFluxInboundEndpoint simpleInboundEndpoint() {
WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint();
RequestMapping requestMapping = new RequestMapping();
requestMapping.setPathPatterns("/test");
endpoint.setRequestMapping(requestMapping);
endpoint.setRequestChannelName("serviceChannel");
return endpoint;
}
@ServiceActivator(inputChannel = "serviceChannel")
String service() {
return "It works!";
}
}
The configuration is similar to the HttpRequestHandlingEndpointSupport
(mentioned prior to the example), except that we use @EnableWebFlux
to add the WebFlux infrastructure to our integration application.
Also, the WebFluxInboundEndpoint
performs sendAndReceive
operations to the downstream flow by using back-pressure, on-demand based capabilities, provided by the reactive HTTP server implementation.
The reply part is non-blocking as well and is based on the internal FutureReplyChannel , which is flat-mapped to a reply Mono for on-demand resolution.
|
You can configure the WebFluxInboundEndpoint
with a custom ServerCodecConfigurer
, a RequestedContentTypeResolver
, and even a ReactiveAdapterRegistry
.
The latter provides a mechanism you can use to return a reply as any reactive type: Reactor Flux
, RxJava Observable
, Flowable
, and others.
This way, we can implement Server Sent Events scenarios with Spring Integration components, as the following example shows:
@Bean
public IntegrationFlow sseFlow() {
return IntegrationFlows
.from(WebFlux.inboundGateway("/sse")
.requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)))
.handle((p, h) -> Flux.just("foo", "bar", "baz"))
.get();
}
See Request Mapping Support and Cross-origin Resource Sharing (CORS) Support for more possible configuration options.
When the request body is empty or payloadExpression
returns null
, the request params (MultiValueMap<String, String>
) is used for a payload
of the target message to process.
35.2. WebFlux Outbound Components
The WebFluxRequestExecutingMessageHandler
(starting with version 5.0) implementation is similar to HttpRequestExecutingMessageHandler
.
It uses a WebClient
from the Spring Framework WebFlux module.
To configure it, define a bean similar to the following:
<bean id="httpReactiveOutbound"
class="org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler">
<constructor-arg value="http://localhost:8080/example" />
<property name="outputChannel" ref="responseChannel" />
</bean>
You can configure a WebClient
instance to use, as the following example shows:
<beans:bean id="webClient" class="org.springframework.web.reactive.function.client.WebClient"
factory-method="create"/>
<bean id="httpReactiveOutbound"
class="org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler">
<constructor-arg value="http://localhost:8080/example" />
<constructor-arg re="webClient" />
<property name="outputChannel" ref="responseChannel" />
</bean>
The WebClient
exchange()
operation returns a Mono<ClientResponse>
, which is mapped (by using several Mono.map()
steps) to an AbstractIntegrationMessageBuilder
as the output from the WebFluxRequestExecutingMessageHandler
.
Together with the ReactiveChannel
as an outputChannel
, the Mono<ClientResponse>
evaluation is deferred until a downstream subscription is made.
Otherwise, it is treated as an async
mode, and the Mono
response is adapted to a SettableListenableFuture
for an asynchronous reply from the WebFluxRequestExecutingMessageHandler
.
The target payload of the output message depends on the WebFluxRequestExecutingMessageHandler
configuration.
The setExpectedResponseType(Class<?>)
or setExpectedResponseTypeExpression(Expression)
identifies the target type of the response body element conversion.
If replyPayloadToFlux
is set to true
, the response body is converted to a Flux
with the provided expectedResponseType
for each element, and this Flux
is sent as the payload downstream.
Afterwards, you can use a splitter to iterate over this Flux
in a reactive manner.
In addition a BodyExtractor<?, ClientHttpResponse>
can be injected into the WebFluxRequestExecutingMessageHandler
instead of the expectedResponseType
and replyPayloadToFlux
properties.
It can be used for low-level access to the ClientHttpResponse
and more control over body and HTTP headers conversion.
Spring Integration provides ClientHttpResponseBodyExtractor
as a identity function to produce (downstream) the whole ClientHttpResponse
and any other possible custom logic.
See HTTP Outbound Components for more possible configuration options.
35.3. WebFlux Namespace Support
Spring Integration provides a webflux
namespace and the corresponding schema definition.
To include it in your configuration, include the following namespace declaration in your application context configuration file:
<?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-webflux="http://www.springframework.org/schema/integration/webflux"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/webflux
https://www.springframework.org/schema/integration/webflux/spring-integration-webflux.xsd">
...
</beans>
35.3.1. Inbound
To configure Spring Integration WebFlux with XML, you caus use appropriate components from the int-webflux
namespace: inbound-channel-adapter
or inbound-gateway
, corresponding to request and response requirements, respectively.
The following example shows how to configure both an inbound channel adapter and an inbound gateway:
<inbound-channel-adapter id="reactiveFullConfig" channel="requests"
path="test1"
auto-startup="false"
phase="101"
request-payload-type="byte[]"
error-channel="errorChannel"
payload-expression="payload"
supported-methods="PUT"
status-code-expression="'202'"
header-mapper="headerMapper"
codec-configurer="codecConfigurer"
reactive-adapter-registry="reactiveAdapterRegistry"
requested-content-type-resolver="requestedContentTypeResolver">
<request-mapping headers="foo"/>
<cross-origin origin="foo"
method="PUT"/>
<header name="foo" expression="'foo'"/>
</inbound-channel-adapter>
<inbound-gateway id="reactiveFullConfig" request-channel="requests"
path="test1"
auto-startup="false"
phase="101"
request-payload-type="byte[]"
error-channel="errorChannel"
payload-expression="payload"
supported-methods="PUT"
reply-timeout-status-code-expression="'504'"
header-mapper="headerMapper"
codec-configurer="codecConfigurer"
reactive-adapter-registry="reactiveAdapterRegistry"
requested-content-type-resolver="requestedContentTypeResolver">
<request-mapping headers="foo"/>
<cross-origin origin="foo"
method="PUT"/>
<header name="foo" expression="'foo'"/>
</inbound-gateway>
35.3.2. Outbound
If you want to execute the HTTP request in a reactive, non-blocking way, you can use the outbound-gateway
or outbound-channel-adapter
.
The following example shows how to configure both an outbound gateway and an outbound channel adapter:
<int-webflux:outbound-gateway id="reactiveExample1"
request-channel="requests"
url="http://localhost/test"
http-method-expression="headers.httpMethod"
extract-request-payload="false"
expected-response-type-expression="payload"
charset="UTF-8"
reply-timeout="1234"
reply-channel="replies"/>
<int-webflux:outbound-channel-adapter id="reactiveExample2"
url="http://localhost/example"
http-method="GET"
channel="requests"
charset="UTF-8"
extract-payload="false"
expected-response-type="java.lang.String"
order="3"
auto-startup="false"/>
35.4. Configuring WebFlux Endpoints with Java
The following example shows how to configure a WebFlux inbound endpoint with Java:
@Bean
public WebFluxInboundEndpoint jsonInboundEndpoint() {
WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint();
RequestMapping requestMapping = new RequestMapping();
requestMapping.setPathPatterns("/persons");
endpoint.setRequestMapping(requestMapping);
endpoint.setRequestChannel(fluxResultChannel());
return endpoint;
}
@Bean
public MessageChannel fluxResultChannel() {
return new FluxMessageChannel();
}
@ServiceActivator(inputChannel = "fluxResultChannel")
Flux<Person> getPersons() {
return Flux.just(new Person("Jane"), new Person("Jason"), new Person("John"));
}
The following example shows how to configure a WebFlux inbound gateway with the Java DSL:
@Bean
public IntegrationFlow inboundChannelAdapterFlow() {
return IntegrationFlows
.from(WebFlux.inboundChannelAdapter("/reactivePost")
.requestMapping(m -> m.methods(HttpMethod.POST))
.requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class))
.statusCodeFunction(m -> HttpStatus.ACCEPTED))
.channel(c -> c.queue("storeChannel"))
.get();
}
The following example shows how to configure a WebFlux outbound gateway with Java:
@ServiceActivator(inputChannel = "reactiveHttpOutRequest")
@Bean
public WebFluxRequestExecutingMessageHandler reactiveOutbound(WebClient client) {
WebFluxRequestExecutingMessageHandler handler =
new WebFluxRequestExecutingMessageHandler("http://localhost:8080/foo", client);
handler.setHttpMethod(HttpMethod.POST);
handler.setExpectedResponseType(String.class);
return handler;
}
The following example shows how to configure a WebFlux outbound gateway with the Java DSL:
@Bean
public IntegrationFlow outboundReactive() {
return f -> f
.handle(WebFlux.<MultiValueMap<String, String>>outboundGateway(m ->
UriComponentsBuilder.fromUriString("http://localhost:8080/foo")
.queryParams(m.getPayload())
.build()
.toUri())
.httpMethod(HttpMethod.GET)
.expectedResponseType(String.class));
}
35.5. WebFlux Header Mappings
Since WebFlux components are fully based on the HTTP protocol, there is no difference in the HTTP headers mapping. See HTTP Header Mappings for more possible options and components to use for mapping headers.
36. WebSockets Support
Starting with version 4.1, Spring Integration has WebSocket support.
It is based on the architecture, infrastructure, and API from the Spring Framework’s web-socket
module.
Therefore, many of Spring WebSocket’s components (such as SubProtocolHandler
or WebSocketClient
) and configuration options (such as @EnableWebSocketMessageBroker
) can be reused within Spring Integration.
For more information, see the Spring Framework WebSocket Support chapter in the Spring Framework reference manual.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-websocket</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-websocket:5.1.5.RELEASE"
For server side, the org.springframework:spring-webmvc
dependency must be included explicitly.
The Spring Framework WebSocket infrastructure is based on the Spring messaging foundation and provides a basic messaging framework based on the same MessageChannel
implementations and MessageHandler
implementations that Spring Integration uses (and some POJO-method annotation mappings).
Consequently, Spring Integration can be directly involved in a WebSocket flow, even without WebSocket adapters.
For this purpose, you can configure a Spring Integration @MessagingGateway
with appropriate annotations, as the following example shows:
@MessagingGateway
@Controller
public interface WebSocketGateway {
@MessageMapping("/greeting")
@SendToUser("/queue/answer")
@Gateway(requestChannel = "greetingChannel")
String greeting(String payload);
}
36.1. Overview
Since the WebSocket protocol is streaming by definition and we can send and receive messages to and from a WebSocket at the same time, we can deal with an appropriate WebSocketSession
, regardless of being on the client or server side.
To encapsulate the connection management and WebSocketSession
registry, the IntegrationWebSocketContainer
is provided with ClientWebSocketContainer
and ServerWebSocketContainer
implementations.
Thanks to the WebSocket API and its implementation in the Spring Framework (with many extensions), the same classes are used on the server side as well as the client side (from a Java perspective, of course).
Consequently, most connection and WebSocketSession
registry options are the same on both sides.
That lets us reuse many configuration items and infrastructure hooks to build WebSocket applications on the server side as well as on the client side.
The following example shows how components can serve both purposes:
//Client side
@Bean
public WebSocketClient webSocketClient() {
return new SockJsClient(Collections.singletonList(new WebSocketTransport(new JettyWebSocketClient())));
}
@Bean
public IntegrationWebSocketContainer clientWebSocketContainer() {
return new ClientWebSocketContainer(webSocketClient(), "ws://my.server.com/endpoint");
}
//Server side
@Bean
public IntegrationWebSocketContainer serverWebSocketContainer() {
return new ServerWebSocketContainer("/endpoint").withSockJs();
}
The IntegrationWebSocketContainer
is designed to achieve bidirectional messaging and can be shared between inbound and outbound channel adapters (see below), can be referenced from only one of them when using one-way (sending or receiving) WebSocket messaging.
It can be used without any channel adapter, but, in this case, IntegrationWebSocketContainer
only plays a role as the WebSocketSession
registry.
The ServerWebSocketContainer implements WebSocketConfigurer to register an internal IntegrationWebSocketContainer.IntegrationWebSocketHandler as an Endpoint .
It does so under the provided paths and other server WebSocket options (such as HandshakeHandler or SockJS fallback ) within the ServletWebSocketHandlerRegistry for the target vendor WebSocket Container.
This registration is achieved with an infrastructural WebSocketIntegrationConfigurationInitializer component, which does the same as the @EnableWebSocket annotation.
This means that, by using @EnableIntegration (or any Spring Integration namespace in the application context), you can omit the @EnableWebSocket declaration, because the Spring Integration infrastructure detects all WebSocket endpoints.
|
36.2. WebSocket Inbound Channel Adapter
The WebSocketInboundChannelAdapter
implements the receiving part of WebSocketSession
interaction.
You must supply it with a IntegrationWebSocketContainer
, and the adapter registers itself as a WebSocketListener
to handle incoming messages and WebSocketSession
events.
Only one WebSocketListener can be registered in the IntegrationWebSocketContainer .
|
For WebSocket subprotocols, the WebSocketInboundChannelAdapter
can be configured with SubProtocolHandlerRegistry
as the second constructor argument.
The adapter delegates to the SubProtocolHandlerRegistry
to determine the appropriate SubProtocolHandler
for the accepted WebSocketSession
and to convert a WebSocketMessage
to a Message
according to the sub-protocol implementation.
By default, the WebSocketInboundChannelAdapter relies only on the raw PassThruSubProtocolHandler implementation, which converts the WebSocketMessage to a Message .
|
The WebSocketInboundChannelAdapter
accepts and sends to the underlying integration flow only Message
instances that have SimpMessageType.MESSAGE
or an empty simpMessageType
header.
All other Message
types are handled through the ApplicationEvent
instances emitted from a SubProtocolHandler
implementation (such as
StompSubProtocolHandler
).
On the server side, if the @EnableWebSocketMessageBroker
configuration is present, you can configure WebSocketInboundChannelAdapter
with the useBroker = true
option.
In this case, all non-MESSAGE
Message
types are delegated to the provided AbstractBrokerMessageHandler
.
In addition, if the broker relay is configured with destination prefixes, those messages that match the Broker destinations are routed to the AbstractBrokerMessageHandler
instead of to the outputChannel
of the WebSocketInboundChannelAdapter
.
If useBroker = false
and the received message is of the SimpMessageType.CONNECT
type, the WebSocketInboundChannelAdapter
immediately sends a SimpMessageType.CONNECT_ACK
message to the WebSocketSession
without sending it to the channel.
Spring’s WebSocket Support allows the configuration of only one broker relay.
Consequently, we do not require an AbstractBrokerMessageHandler reference.
It is detected in the Application Context.
|
For more configuration options, see WebSockets Namespace Support.
36.3. WebSocket Outbound Channel Adapter
The WebSocketOutboundChannelAdapter
:
-
Accepts Spring Integration messages from its
MessageChannel
-
Determines the
WebSocketSession
id
from theMessageHeaders
-
Retrieves the
WebSocketSession
from the providedIntegrationWebSocketContainer
-
Delegates the conversion and sending of
WebSocketMessage
work to the appropriateSubProtocolHandler
from the providedSubProtocolHandlerRegistry
.
On the client side, the WebSocketSession
id
message header is not required, because ClientWebSocketContainer
deals only with a single connection and its WebSocketSession
respectively.
To use the STOMP sub-protocol, you should configure this adapter with a StompSubProtocolHandler
.
Then you can send any STOMP message type to this adapter, using StompHeaderAccessor.create(StompCommand…)
and a MessageBuilder
, or just using a HeaderEnricher
(see Header Enricher).
The rest of this chapter covers largely additional configuration options.
36.4. WebSockets Namespace Support
The Spring Integration WebSocket namespace includes several components described in the remainder of this chapter. To include it in your configuration, use the following namespace declaration in your application context configuration file:
<?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-websocket="http://www.springframework.org/schema/integration/websocket"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/websocket
https://www.springframework.org/schema/integration/websocket/spring-integration-websocket.xsd">
...
</beans>
36.4.1. <int-websocket:client-container>
Attributes
The following listing shows the attributes available for the <int-websocket:client-container>
element:
<int-websocket:client-container
id="" (1)
client="" (2)
uri="" (3)
uri-variables="" (4)
origin="" (5)
send-time-limit="" (6)
send-buffer-size-limit="" (7)
auto-startup="" (8)
phase=""> (9)
<int-websocket:http-headers>
<entry key="" value=""/>
</int-websocket:http-headers> (10)
</int-websocket:client-container>
| 1 | The component bean name. |
| 2 | The WebSocketClient bean reference. |
| 3 | The uri or uriTemplate to the target WebSocket service.
If you use it as a uriTemplate with URI variable placeholders, the uri-variables attribute is required. |
| 4 | Comma-separated values for the URI variable placeholders within the uri attribute value.
The values are replaced into the placeholders according to their order in the uri .
See UriComponents.expand(Object…uriVariableValues) . |
| 5 | The Origin Handshake HTTP header value. |
| 6 | The WebSocket session 'send' timeout limit.
Defaults to 10000 . |
| 7 | The WebSocket session 'send' message size limit.
Defaults to 524288 . |
| 8 | Boolean value indicating whether this endpoint should start automatically.
Defaults to false , assuming that this container is started from the WebSocket inbound adapter. |
| 9 | The lifecycle phase within which this endpoint should start and stop.
The lower the value, the earlier this endpoint starts and the later it stops.
The default is Integer.MAX_VALUE .
Values can be negative.
See SmartLifeCycle . |
| 10 | A Map of HttpHeaders to be used with the Handshake request. |
36.4.2. <int-websocket:server-container>
Attributes
The following listing shows the attributes available for the <int-websocket:server-container>
element:
<int-websocket:server-container
id="" (1)
path="" (2)
handshake-handler="" (3)
handshake-interceptors="" (4)
decorator-factories="" (5)
send-time-limit="" (6)
send-buffer-size-limit="" (7)
allowed-origins=""> (8)
<int-websocket:sockjs
client-library-url="" (9)
stream-bytes-limit="" (10)
session-cookie-needed="" (11)
heartbeat-time="" (12)
disconnect-delay="" (13)
message-cache-size="" (14)
websocket-enabled="" (15)
scheduler="" (16)
message-codec="" (17)
transport-handlers="" (18)
suppress-cors="true"="" /> (19)
</int-websocket:server-container>
| 1 | The component bean name. |
| 2 | A path (or comma-separated paths) that maps a particular request to a WebSocketHandler .
Supports exact path mapping URIs (such as /myPath ) and ant-style path patterns (such as /myPath/** ). |
| 3 | The HandshakeHandler bean reference.
Defaults to DefaultHandshakeHandler . |
| 4 | List of HandshakeInterceptor bean references. |
| 5 | List of one or more factories (WebSocketHandlerDecoratorFactory ) that decorate the handler used to process WebSocket messages.
This may be useful for some advanced use cases (for example, to allow Spring Security to forcibly close
the WebSocket session when the corresponding HTTP session expires).
See the Spring Session Project for more information. |
| 6 | See the same option on the <int-websocket:client-container> . |
| 7 | See the same option on the <int-websocket:client-container> . |
| 8 | The allowed origin header values.
You can specify multiple origins as a comma-separated list.
This check is mostly designed for browser clients.
There is nothing preventing other types of client from modifying the origin header value.
When SockJS is enabled and allowed origins are restricted, transport types that do not use origin headers for cross-origin requests (jsonp-polling , iframe-xhr-polling , iframe-eventsource , and iframe-htmlfile ) are disabled.
As a consequence, IE6 and IE7 are not supported, and IE8 and IE9 are supported only without cookies.
By default, all origins are allowed. |
| 9 | Transports with no native cross-domain communication (such as eventsource and htmlfile ) must get a simple page from the “foreign” domain in an invisible iframe so that code in the iframe can run from a domain local to the SockJS server.
Since the iframe needs to load the SockJS javascript client library, this property lets you specify the location from which to load it.
By default, it points to d1fxtkz8shb9d2.cloudfront.net/sockjs-0.3.4.min.js .
However, you can also set it to point to a URL served by the application.
Note that it is possible to specify a relative URL, in which case the URL must be relative to the iframe URL.
For example, assuming a SockJS endpoint mapped to /sockjs and the resulting iframe URL is /sockjs/iframe.html , the relative URL must start with "../../" to traverse up to the location above the SockJS mapping.
For prefix-based servlet mapping, you may need one more traversal. |
| 10 | Minimum number of bytes that can be sent over a single HTTP streaming request before it is closed.
Defaults to 128K (that is, 128*1024 or 131072 bytes). |
| 11 | The cookie_needed value in the response from the SockJs /info endpoint.
This property indicates whether a JSESSIONID cookie is required for the application to function correctly (for example, for load balancing or in Java Servlet containers for the use of an HTTP session). |
| 12 | The amount of time (in milliseconds) when the server has not sent any messages and after which the server should
send a heartbeat frame to the client in order to keep the connection from breaking.
The default value is 25,000 (25 seconds). |
| 13 | The amount of time (in milliseconds) before a client is considered disconnected after not having a receiving connection (that is, an active connection over which the server can send data to the client).
The default value is 5000 . |
| 14 | The number of server-to-client messages that a session can cache while waiting for the next HTTP polling request from the client.
The default size is 100 . |
| 15 | Some load balancers do not support WebSockets.
Set this option to false to disable the WebSocket transport on the server side.
The default value is true . |
| 16 | The TaskScheduler bean reference.
A new ThreadPoolTaskScheduler instance is created if no value is provided.
This scheduler instance is used for scheduling heart-beat messages. |
| 17 | The SockJsMessageCodec bean reference to use for encoding and decoding SockJS messages.
By default, Jackson2SockJsMessageCodec is used, which requires the Jackson library to be present on the classpath. |
| 18 | List of TransportHandler bean references. |
| 19 | Whether to disable automatic addition of CORS headers for SockJS requests.
The default value is false . |
36.4.3. <int-websocket:outbound-channel-adapter>
Attributes
The following listing shows the attributes available for the <int-websocket:outbound-channel-adapter>
element:
<int-websocket:outbound-channel-adapter
id="" (1)
channel="" (2)
container="" (3)
default-protocol-handler="" (4)
protocol-handlers="" (5)
message-converters="" (6)
merge-with-default-converters="" (7)
auto-startup="" (8)
phase=""/> (9)
| 1 | The component bean name.
If you do not provide the channel attribute, a DirectChannel is created and registered in the application context with this id attribute as the bean name.
In this case, the endpoint is registered with the bean name id plus .adapter .
And the MessageHandler is registered with the bean alias id plus .handler . |
| 2 | Identifies the channel attached to this adapter. |
| 3 | The reference to the IntegrationWebSocketContainer bean, which encapsulates the low-level connection and WebSocketSession handling operations.
Required. |
| 4 | Optional reference to a SubProtocolHandler instance.
It is used when the client did not request a sub-protocol or it is a single protocol-handler.
If this reference or a protocol-handlers list is not provided, the PassThruSubProtocolHandler is used by default. |
| 5 | List of SubProtocolHandler bean references for this channel adapter.
If you provide only a single bean reference and do not provide a default-protocol-handler , that single SubProtocolHandler is used as the default-protocol-handler .
If you do not set this attribute or default-protocol-handler , the PassThruSubProtocolHandler is used by default. |
| 6 | List of MessageConverter bean references for this channel adapter. |
| 7 | Boolean value indicating whether the default converters should be registered after any custom converters.
This flag is used only if message-converters is provided.
Otherwise, all default converters are registered.
Defaults to false .
The default converters are (in order): StringMessageConverter , ByteArrayMessageConverter , and MappingJackson2MessageConverter (if the Jackson library is present on the classpath). |
| 8 | Boolean value indicating whether this endpoint should start automatically.
Defaults to true . |
| 9 | The lifecycle phase within which this endpoint should start and stop.
The lower the value, the earlier this endpoint starts and the later it stops.
The default is Integer.MIN_VALUE .
Values can be negative.
See SmartLifeCycle . |
36.4.4. <int-websocket:inbound-channel-adapter>
Attributes
The following listing shows the attributes available for the <int-websocket:outbound-channel-adapter>
element:
<int-websocket:inbound-channel-adapter
id="" (1)
channel="" (2)
error-channel="" (3)
container="" (4)
default-protocol-handler="" (5)
protocol-handlers="" (6)
message-converters="" (7)
merge-with-default-converters="" (8)
send-timeout="" (9)
payload-type="" (10)
use-broker="" (11)
auto-startup="" (12)
phase=""/> (13)
| 1 | The component bean name.
If you do not set the channel attribute, a DirectChannel is created and registered in the application context with this id attribute as the bean name.
In this case, the endpoint is registered with the bean name id plus .adapter . |
| 2 | Identifies the channel attached to this adapter. |
| 3 | The MessageChannel bean reference to which the ErrorMessage instances should be sent. |
| 4 | See the same option on the <int-websocket:outbound-channel-adapter> . |
| 5 | See the same option on the <int-websocket:outbound-channel-adapter> . |
| 6 | See the same option on the <int-websocket:outbound-channel-adapter> . |
| 7 | See the same option on the <int-websocket:outbound-channel-adapter> . |
| 8 | See the same option on the <int-websocket:outbound-channel-adapter> . |
| 9 | Maximum amount of time (in milliseconds) to wait when sending a message to the channel if the channel can block.
For example, a QueueChannel can block until space is available if its maximum capacity has been reached. |
| 10 | Fully qualified name of the Java type for the target payload to convert from the incoming WebSocketMessage .
Defaults to java.lang.String . |
| 11 | Indicates whether this adapter sends non-MESSAGE WebSocketMessage instances and messages with broker destinations to the AbstractBrokerMessageHandler from the application context.
When this attribute is true , the Broker Relay configuration is required.
This attribute is used only on the server side.
On the client side, it is ignored.
Defaults to false . |
| 12 | See the same option on the <int-websocket:outbound-channel-adapter> . |
| 13 | See the same option on the <int-websocket:outbound-channel-adapter> . |
36.5. Using ClientStompEncoder
Starting with version 4.3.13, Spring Integration provides ClientStompEncoder
(as an extension of the standard StompEncoder
) for use on the client side of WebSocket channel adapters.
For proper client side message preparation, you must inject an instance of the ClientStompEncoder
into the StompSubProtocolHandler
.
One problem with the default StompSubProtocolHandler
is that it was designed for the server side, so it updates the SEND
stompCommand
header into MESSAGE
(as required by the STOMP protocol for the server side).
If the client does not send its messages in the proper SEND
web socket frame, some STOMP brokers do not accept them.
The purpose of the ClientStompEncoder
, in this case, is to override the stompCommand
header and set it to the SEND
value before encoding the message to the byte[]
.
37. Web Services Support
This chapter describes Spring Integration’s support for web services, including:
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ws</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-ws:5.1.5.RELEASE"
37.1. Outbound Web Service Gateways
To invoke a web service when you send a message to a channel, you have two options, both of which build upon the Spring Web Services project: SimpleWebServiceOutboundGateway
and MarshallingWebServiceOutboundGateway
.
The former accepts either a String
or javax.xml.transform.Source
as the message payload.
The latter supports any implementation of the Marshaller
and Unmarshaller
interfaces.
Both require a Spring Web Services DestinationProvider
, to determine the URI of the web service to be called.
The following example shows both options for invoking a web service:
simpleGateway = new SimpleWebServiceOutboundGateway(destinationProvider);
marshallingGateway = new MarshallingWebServiceOutboundGateway(destinationProvider, marshaller);
When using the namespace support (described later), you need only set a URI.
Internally, the parser configures a fixed URI DestinationProvider implementation.
If you need dynamic resolution of the URI at runtime, however, then the DestinationProvider can provide such behavior as looking up the URI from a registry.
See the Spring Web Services DestinationProvider Javadoc for more information about this strategy.
|
Starting with version 5.0, you can supply the SimpleWebServiceOutboundGateway
and MarshallingWebServiceOutboundGateway
with an external WebServiceTemplate
instance, which you can configure for any custom properties, including checkConnectionForFault
(which allows your application to deal with non-conforming services).
For more detail on the inner workings, see the Spring Web Services reference guide’s chapter covering client access and the chapter covering Object/XML mapping.
37.2. Inbound Web Service Gateways
To send a message to a channel upon receiving a web service invocation, you again have two options: SimpleWebServiceInboundGateway
and MarshallingWebServiceInboundGateway
.
The former extracts a javax.xml.transform.Source
from the WebServiceMessage
and sets it as the message payload.
The latter supports implementation of the Marshaller
and Unmarshaller
interfaces.
If the incoming web service message is a SOAP message, the SOAP action header is added to the headers of the Message
that is forwarded onto the request channel.
The following example shows both options:
simpleGateway = new SimpleWebServiceInboundGateway();
simpleGateway.setRequestChannel(forwardOntoThisChannel);
simpleGateway.setReplyChannel(listenForResponseHere); //Optional
marshallingGateway = new MarshallingWebServiceInboundGateway(marshaller);
//set request and optionally reply channel
Both gateways implement the Spring Web Services MessageEndpoint
interface, so they can be configured with a MessageDispatcherServlet
as per standard Spring Web Services configuration.
For more detail on how to use these components, see the Spring Web Services reference guide’s chapter covering creating a web service. The chapter covering Object/XML mapping is also applicable again.
To add the SimpleWebServiceInboundGateway
and MarshallingWebServiceInboundGateway
configurations to the Spring WS infrastructure, you should add the EndpointMapping
definition between MessageDispatcherServlet
and the target MessageEndpoint
implementations, as you would for a normal Spring WS application.
For this purpose (from the Spring Integration perspective), Spring WS provides the following convenient EndpointMapping
implementations:
-
o.s.ws.server.endpoint.mapping.UriEndpointMapping
-
o.s.ws.server.endpoint.mapping.PayloadRootQNameEndpointMapping
-
o.s.ws.soap.server.endpoint.mapping.SoapActionEndpointMapping
-
o.s.ws.server.endpoint.mapping.XPathPayloadEndpointMapping
You must specify the beans for these classes in the application context and reference the SimpleWebServiceInboundGateway
and/or MarshallingWebServiceInboundGateway
bean definitions according to the WS mapping algorithm.
See the endpoint mappings for more information.
37.3. Web Service Namespace Support
To configure an outbound web service gateway, use the outbound-gateway
element from the ws
namespace, as the following example shows:
<int-ws:outbound-gateway id="simpleGateway"
request-channel="inputChannel"
uri="https://example.org"/>
This example does not provide a 'reply-channel'.
If the web service returns a non-empty response, the Message containing that response is sent to the reply channel defined in the request message’s REPLY_CHANNEL header.
If that is not available, a channel resolution exception is thrown.
If you want to send the reply to another channel instead, provide a 'reply-channel' attribute on the 'outbound-gateway' element.
|
By default, when you invoke a web service that returns an empty response after using a String payload for the request Message , no reply Message is sent.
Therefore, you need not set a 'reply-channel' or have a REPLY_CHANNEL header in the request Message .
If you actually do want to receive the empty response as a Message , you can set the 'ignore-empty-responses' attribute to false .
Doing so works only for String objects, because using a Source or a Document object leads to a null response and consequently never generates a reply Message .
|
To set up an inbound Web Service Gateway, use the inbound-gateway
element, as the following example shows:
<int-ws:inbound-gateway id="simpleGateway"
request-channel="inputChannel"/>
To use Spring OXM marshallers or unmarshallers, you must provide bean references. The following example shows how to provide a bean reference for an outbound marshalling gateway:
<int-ws:outbound-gateway id="marshallingGateway"
request-channel="requestChannel"
uri="https://example.org"
marshaller="someMarshaller"
unmarshaller="someUnmarshaller"/>
The following example shows how to provide a bean reference for an inbound marshalling gateway:
<int-ws:inbound-gateway id="marshallingGateway"
request-channel="requestChannel"
marshaller="someMarshaller"
unmarshaller="someUnmarshaller"/>
Most Marshaller implementations also implement the Unmarshaller interface.
When using such a Marshaller , only the marshaller attribute is necessary.