To configure STOMP, you should start with the STOMP client object. The Spring Framework provides the following implementations:

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

<int-stomp:outbound-channel-adapter
                           id=""                      
                           channel=""                 
                           stomp-session-manager=""   
                           header-mapper=""           
                           mapped-headers=""          
                           destination=""             
                           destination-expression=""  
                           auto-startup=""            
                           phase=""/>                 

<int-stomp:inbound-channel-adapter
                           id=""                     
                           channel=""                
                           error-channel=""          
                           stomp-session-manager=""  
                           header-mapper=""          
                           mapped-headers=""         
                           destinations=""           
                           send-timeout=""           
                           payload-type=""           
                           auto-startup=""           
                           phase=""/>