Spring Integration version 4.2 introduced STOMP (Simple Text Orientated Messaging Protocol) client support.
It is based on the architecture, infrastructure, and API from the Spring Framework’s messaging module, stomp package.
Spring Integration uses many of Spring STOMP components (such as StompSession
and StompClientSupport
).
For more information, see the Spring Framework STOMP Support chapter in the Spring Framework reference manual.
You need to include this dependency into your project:
Maven.
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-stomp</artifactId> <version>5.1.0.RELEASE</version> </dependency>
Gradle.
compile "org.springframework.integration:spring-integration-stomp:5.1.0.RELEASE"
For server side components you need to add a org.springframework:spring-websocket
and/or io.projectreactor.netty:reactor-netty
dependencies.
To configure STOMP, you should start with the STOMP client object. The Spring Framework provides the following implementations:
WebSocketStompClient
: Built on the Spring WebSocket API with support for standard JSR-356 WebSocket, Jetty 9, and SockJS for HTTP-based WebSocket emulation with SockJS Client.
ReactorNettyTcpStompClient
: Built on ReactorNettyTcpClient
from the reactor-netty
project.
You can provide any other StompClientSupport
implementation.
See the Javadoc of those classes.
The StompClientSupport
class is designed as a factory to produce a StompSession
for the provided StompSessionHandler
and all the remaining work is done through the callbacks to that StompSessionHandler
and StompSession
abstraction.
With the Spring Integration adapter abstraction, we need to provide some managed shared object to represent our application as a STOMP client with its unique session.
For this purpose, Spring Integration provides the StompSessionManager
abstraction to manage the single StompSession
between any provided StompSessionHandler
.
This allows the use of inbound or outbound channel adapters (or both) for the particular STOMP Broker.
See StompSessionManager
(and its implementations) JavaDocs for more information.
The StompInboundChannelAdapter
is a one-stop MessageProducer
component that subscribes your Spring Integration application to the provided STOMP destinations and receives messages from them (converted from the STOMP frames by using the provided MessageConverter
on the connected StompSession
).
You can change the destinations (and therefore STOMP subscriptions) at runtime by using appropriate @ManagedOperation
annotations on the StompInboundChannelAdapter
.
For more configuration options, see Section 31.7, “STOMP Namespace Support” and the StompInboundChannelAdapter
Javadoc.
The StompMessageHandler
is the MessageHandler
for the <int-stomp:outbound-channel-adapter>
and is used to send the outgoing Message<?>
instances to the STOMP destination
(pre-configured or determined at runtime with a SpEL expression) through the StompSession
(which is provided by the shared StompSessionManager
).
For more configuration options see Section 31.7, “STOMP Namespace Support” and the StompMessageHandler
Javadoc.
The STOMP protocol provides headers as part of its frame. The entire structure of the STOMP frame has the following format:
.... COMMAND header1:value1 header2:value2 Body^@ ....
Spring Framework provides StompHeaders
to represent these headers.
See the Javadoc for more details.
STOMP frames are converted to and from Message<?>
instances and these headers are mapped to and from MessageHeaders
instances.
Spring Integration provides a default HeaderMapper
implementation for the STOMP adapters.
The implementation is StompHeaderMapper
.
It provides fromHeaders()
and toHeaders()
operations for the inbound and outbound adapters, respectively.
As with many other Spring Integration modules, the IntegrationStompHeaders
class has been introduced to map standard STOMP headers to MessageHeaders
, with stomp_
as the header name prefix.
In addition, all MessageHeaders
instances with that prefix are mapped to the StompHeaders
when sending to a destination.
For more information, see the Javadoc for those classes and the mapped-headers
attribute description in the Section 31.7, “STOMP Namespace Support”.
Many STOMP operations are asynchronous, including error handling.
For example, STOMP has a RECEIPT
server frame that it returns when a client frame has requested one by adding the RECEIPT
header.
To provide access to these asynchronous events, Spring Integration emits StompIntegrationEvent
instances, which you can obtain by implementing an ApplicationListener
or by using an <int-event:inbound-channel-adapter>
(see Section 15.1, “Receiving Spring Application Events”).
Specifically, a StompExceptionEvent
is emitted from the AbstractStompSessionManager
when a stompSessionListenableFuture
receives onFailure()
due to failure to connect to STOMP broker.
Another example is the StompMessageHandler
.
It processes ERROR
STOMP frames, which are server responses to improper (unaccepted) messages sent by this StompMessageHandler
.
The StompMessageHandler
emits StompReceiptEvent
as a part of StompSession.Receiptable
callbacks in the asynchronous answers for the messages sent to the StompSession
.
The StompReceiptEvent
can be positive or negative, depending on whether or not the RECEIPT
frame was received from the server within the receiptTimeLimit
period, which you can configure on the StompClientSupport
instance.
It defaults to 15 * 1000
(in milliseconds, so 15 seconds).
Note | |
---|---|
The |
See Section 31.6, “STOMP Adapters Java Configuration” for more information how to configure Spring Integration to accept those ApplicationEvent
instances.
The following example shows a comprehensive Java configuration for STOMP adapters:
@Configuration @EnableIntegration public class StompConfiguration { @Bean public ReactorNettyTcpStompClient stompClient() { ReactorNettyTcpStompClient stompClient = new ReactorNettyTcpStompClient("127.0.0.1", 61613); stompClient.setMessageConverter(new PassThruMessageConverter()); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); stompClient.setTaskScheduler(taskScheduler); stompClient.setReceiptTimeLimit(5000); return stompClient; } @Bean public StompSessionManager stompSessionManager() { ReactorNettyTcpStompSessionManager stompSessionManager = new ReactorNettyTcpStompSessionManager(stompClient()); stompSessionManager.setAutoReceipt(true); return stompSessionManager; } @Bean public PollableChannel stompInputChannel() { return new QueueChannel(); } @Bean public StompInboundChannelAdapter stompInboundChannelAdapter() { StompInboundChannelAdapter adapter = new StompInboundChannelAdapter(stompSessionManager(), "/topic/myTopic"); adapter.setOutputChannel(stompInputChannel()); return adapter; } @Bean @ServiceActivator(inputChannel = "stompOutputChannel") public MessageHandler stompMessageHandler() { StompMessageHandler handler = new StompMessageHandler(stompSessionManager()); handler.setDestination("/topic/myTopic"); return handler; } @Bean public PollableChannel stompEvents() { return new QueueChannel(); } @Bean public ApplicationListener<ApplicationEvent> stompEventListener() { ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer(); producer.setEventTypes(StompIntegrationEvent.class); producer.setOutputChannel(stompEvents()); return producer; } }
The Spring Integration STOMP namespace implements the inbound and outbound channel adapter components. To include it in your configuration, provide 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-stomp="http://www.springframework.org/schema/integration/stomp" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/integration/stomp http://www.springframework.org/schema/integration/stomp/spring-integration-stomp.xsd"> ... </beans>
The following listing shows the available attributes for the STOMP outbound channel adapter:
<int-stomp:outbound-channel-adapter id="" channel="" stomp-session-manager="" header-mapper="" mapped-headers="" destination="" destination-expression="" auto-startup="" phase=""/>
The component bean name.
The | |
Identifies the channel attached to this adapter if | |
Reference to a | |
Reference to a bean that implements | |
Comma-separated list of names of STOMP Headers to be mapped to the STOMP frame headers.
It can be provided only if the | |
Name of the destination to which STOMP Messages are sent.
It is mutually exclusive with the | |
A SpEL expression to be evaluated at runtime against each Spring Integration | |
Boolean value indicating whether this endpoint should start automatically.
It defaults to | |
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 |
The following listing shows the available attributes for the STOMP inbound channel adapter:
<int-stomp:inbound-channel-adapter id="" channel="" error-channel="" stomp-session-manager="" header-mapper="" mapped-headers="" destinations="" send-timeout="" payload-type="" auto-startup="" phase=""/>
The component bean name.
If you do not set the | |
Identifies the channel attached to this adapter. | |
The | |
See the same option on the | |
Comma-separated list of names of STOMP Headers to be mapped from the STOMP frame headers.
You can only provide this if the | |
See the same option on the | |
Comma-separated list of STOMP destination names to subscribe.
The list of destinations (and therefore subscriptions) can be modified at runtime through the | |
Maximum amount of time (in milliseconds) to wait when sending a message to the channel if the channel can block.
For example, a | |
Fully qualified name of the Java type for the target | |
See the same option on the | |
See the same option on the |