STOMP Support

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.4.4</version>
</dependency>

Gradle

compile "org.springframework.integration:spring-integration-stomp:5.4.4"

For server side components you need to add a org.springframework:spring-websocket and/or io.projectreactor.netty:reactor-netty dependencies.

Overview

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.

STOMP Inbound Channel Adapter

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 STOMP Namespace Support and the StompInboundChannelAdapter Javadoc.

STOMP Outbound Channel Adapter

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 STOMP Namespace Support and the StompMessageHandler Javadoc.

STOMP Headers Mapping

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 STOMP Namespace Support.

STOMP Integration Events

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

The StompSession.Receiptable callbacks are added only if the RECEIPT STOMP header of the message to send is not null. You can enable automatic RECEIPT header generation on the StompSession through its autoReceipt option and on the StompSessionManager respectively.

See STOMP Adapters Java Configuration for more information how to configure Spring Integration to accept those ApplicationEvent instances.

STOMP Adapters Java Configuration

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;
    }

}

STOMP Namespace Support

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
    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/stomp
    https://www.springframework.org/schema/integration/stomp/spring-integration-stomp.xsd">
    ...
</beans>

Understanding the `<int-stomp:outbound-channel-adapter>` Element

The following listing shows the available attributes for the STOMP outbound channel adapter:

<int-stomp:outbound-channel-adapter
                           id=""                      (1)
                           channel=""                 (2)
                           stomp-session-manager=""   (3)
                           header-mapper=""           (4)
                           mapped-headers=""          (5)
                           destination=""             (6)
                           destination-expression=""  (7)
                           auto-startup=""            (8)
                           phase=""/>                 (9)

1	The component bean name. The `MessageHandler` is registered with a bean alias of `id` plus `.handler`. 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 a bean name `id` plus `.adapter`.
2	Identifies the channel attached to this adapter if `id` is present. See `id`. Optional.
3	Reference to a `StompSessionManager` bean, which encapsulates the low-level connection and `StompSession` handling operations. Required.
4	Reference to a bean that implements `HeaderMapper<StompHeaders>`, which maps Spring Integration `MessageHeaders` to and from STOMP frame headers. It is mutually exclusive with `mapped-headers`. It defaults to `StompHeaderMapper`.
5	Comma-separated list of names of STOMP Headers to be mapped to the STOMP frame headers. It can be provided only 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 (such as `myheader` or `myheader`). A special token (`STOMP_OUTBOUND_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 include this token or provide your own `HeaderMapper` implementation by using `header-mapper`.
6	Name of the destination to which STOMP Messages are sent. It is mutually exclusive with the `destination-expression`.
7	A SpEL expression to be evaluated at runtime against each Spring Integration `Message` as the root object. It is mutually exclusive with the `destination`.
8	Boolean value indicating whether this endpoint should start automatically. It 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`.

Understanding the `<int-stomp:inbound-channel-adapter>` Element

The following listing shows the available attributes for the STOMP inbound channel adapter:

<int-stomp:inbound-channel-adapter
                           id=""                     (1)
                           channel=""                (2)
                           error-channel=""          (3)
                           stomp-session-manager=""  (4)
                           header-mapper=""          (5)
                           mapped-headers=""         (6)
                           destinations=""           (7)
                           send-timeout=""           (8)
                           payload-type=""           (9)
                           auto-startup=""           (10)
                           phase=""/>                (11)

1	The component bean name. 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>`.