Implementing a STOMP client in a Spring Boot application

Posted on Mar 13, 2022

Note: This article was written a while ago and may contain outdated information. Please verify the details before relying on it. If I express opinions or recommendations, they might not reflect my current views. For this reason, I recommend checking for more recent articles on the same topic.

STOMP is a good and easy protocol to communicate in realtime via WebSockets.

If you want to implement such a client in your Spring application, you can use Springs simp package, which provides all you need to start quickly.

Requirements

This article focuses on the STOMP client, so you need a server you can subscribe to and consume messages from. (See also: A simple STOMP server & producer application)

Also, you need the spring-boot-starter-websocket library as a dependency. E.g. in Gradle:

implementation "org.springframework.boot:spring-boot-starter-websocket:3.2.0-SNAPSHOT"

The channel you subscribe to provides messages in a specific format. These messages will be parsed into POJOs like this:

public class StompMessage {
    String message;

    public StompMessage() {}

    public StompMessage(String message) {
        this.message = message;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }
}

Note: I’m using jackson for deserialization and this needs an empty, default constructor.

Creating the STOMP client

The client acts as a session manager and message handler (which is called a frame). This is achieved by implementing the StompSessionHandler.

You could also implement two different classes for the STOMP session handler and the frame (= message) handler by using these interfaces:

extends StompSessionHandlerAdapter

and

implements StompFrameHandler

The following example has been implemented in Spring Boot 3. There was an older implementation with Spring Boot 2.x. and an older version of the websocket library. You can see it here.

The code implements methods to connect to a server, to subscribe to topics, to unsubscribe or to check the connection state or subscription state. The comments give some hints how the class can be further optimized.

package de.dkwr.stomp;

import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
import org.springframework.stereotype.Controller;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.messaging.WebSocketStompClient;

import java.lang.reflect.Type;

/**
 * Controller for the connection to the STOMP server.
 */
@Controller
@RequiredArgsConstructor
public class StompController implements StompSessionHandler {
    private StompSession stompSession;

    @EventListener(value = ApplicationReadyEvent.class)
    public void connect() {
        WebSocketClient client = new StandardWebSocketClient();
        WebSocketStompClient stompClient = new WebSocketStompClient(client);
        // alternative: stompClient.setMessageConverter(new StringMessageConverter());
        stompClient.setMessageConverter(new MappingJackson2MessageConverter());
        try {
            stompSession = stompClient
                    .connectAsync("ws://localhost:61613/ws", this)
                    .get();
        } catch (Exception e) {
            System.err.println("Connection failed." + e.getMessage()); // Do some failover and implement retry patterns.
        }
    }

    @Override
    public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
        System.err.println("Connection to STOMP server established.\n" +
                "Session: " + session + "\n" +
                "Headers: " + connectedHeaders + "\n");

        subscribe("1");
    }

    /**
     * Subscribes to the feed of the topic.
     *
     * @param topicId The id of the topic to subscribe for.
     */
    public void subscribe(String topicId) {
        System.err.println("Subscribing to topic:  " + topicId);
        stompSession.subscribe("/topic/" + topicId, this);
    }

    @Override
    public void handleFrame(StompHeaders headers, Object payload) {
        StompMessage p = (StompMessage) payload;
        System.err.println("Received message:" + p.getMessage());
    }

    @Override
    public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, Throwable exception) {
    }

    @Override
    public void handleTransportError(StompSession session, Throwable exception) {
        System.err.println("Retrieved a transport error: {}" + exception.getMessage());
        if (!session.isConnected()) {
            connect();
        }
    }

    @Override
    public Type getPayloadType(StompHeaders headers) {
        return StompMessage.class;
    }

    /**
     * Unsubscribe and close connection before destroying this instance (e.g. on application shutdown).
     */
    @PreDestroy
    void onShutDown() {
        if (stompSession != null) {
            stompSession.disconnect();
        }
    }
}

Implementing a STOMP client in a Spring Boot application

Requirements

Creating the STOMP client

Further reading

Want to know more?