Being STOMP Agnostic — Domain Specific Language

1. What is STOMP
2. The search for a Rust STOMP crate that fits the bill
3. Finding a fork
4. How do we fork?
5. Designing an agnostic interface
   1. ClientTransport
   2. ClientStompHandle
6. Implementing a WebSocket client transport
   1. A note on cancel safety
7. Sending and receiving messages
8. What now?

I needed a STOMP library, and my search for one written in Rust ended up with a fork and a library that I'm happy with!

In this blog post I will try to do the following things

• Introduce you to STOMP
• Go through the current landscape of Rust STOMP libraries and explain why none of them scratch my itch
• Find a suitable library to fork, so that we may stand on the shoulders of others
• Propose a runtime and transport agnostic design for a STOMP library and implement it as stomp-agnostic

What is STOMP

STOMP stands for Simple Text Oriented Messaging Protocol. If you have a need for sending and receiving text messages in real time using a client-server model, and you don't have fancy needs, this is the protocol for you. STOMP has been around for a long time; the latest version, STOMP 1.2, was published in 2012.

The protocol itself is simple and easy-to-understand, and is usually used together with the WebSocket communication protocol in browsers, but can be used over TCP or any other protocol as the transport layer. STOMP is supported by a lot of software: RabbitMQ, Spring Framework, and is also commonly used to support real-time updates in web applications.

Transport layer is used a bit lax in the previous section; in the OSI model, TCP is at the Transport layer (layer 4), while HTTP and WebSocket are at the Application layer (layer 7), the same as STOMP. However, when we are in a browser we don't have direct access to TCP/layer 4, so we transport the STOMP data using a WebSocket connection.

Communicating using STOMP is done using Frames. A frame is a verb plus some data; similar to HTTP verbs, but with a broader vocabulary. There are two connection frames: CONNECT and CONNECTED, nine standard client frames: SEND, SUBSCRIBE, UNSUBSCRIBE, ACK, NACK, BEGIN, COMMIT, ABORT and DISCONNECT, and three standard server frames: MESSAGE, RECEIPT, and ERROR.

Using the client and server frames, it's possible to build a robust communication channel between a client and a server. A simple conversation would go something like this:

Client                                                          Server     
                  
  │  CONNECT (accept-version=1.1,1.2, host:stomp.zkc.se, ...)      │ 
  ├───────────────────────────────────────────────────────────────►│ 
  │  CONNECTED (version=1.2, session=xyz, ... )                    │ 
  │◄───────────────────────────────────────────────────────────────┤
  |                                                                | 
  |                             ...                                | 
  |                                                                | 
  |                                                                | 
  │  SUBSCRIBE (id=sub-0, destination=/dest, ...)                  │ 
  ├───────────────────────────────────────────────────────────────►│ 
  |                                                                | 
  |                             ...                                | 
  |                                                                | 
  |                                                                | 
  │  MESSAGE (subscription=sub-0, message-id=0, destination=/dest, │ 
  │           data=JSON {...})                                     │ 
  │◄───────────────────────────────────────────────────────────────┤ 
  │  MESSAGE (subscription=sub-0, message-id=1, destination=/dest, │ 
  │           data=JSON {...})                                     │ 
  │◄───────────────────────────────────────────────────────────────┤ 
  |                                                                | 
  |                             ...                                | 
  │                                                                │ 

Since the protocol is so simple and each frame type typically only requires a few pieces of data (e.g. SUBSCRIBE only needs an id and a destination) it's very easy to wrap your head around the whole thing in a very short time. Just click the link above and start reading. It's not harder than that, really.

The search for a Rust STOMP crate that fits the bill

What I need: A library, written in Rust, that I can use to implement several clients using WebSocket as the transport layer, but with wildly different platforms: Desktop & Mobile applications have access to real operating systems and have good support for Tokio, but I also need support for Web/WASM that has a hard time running Tokio. Web browsers have their own APIs for opening WebSockets. I also need to implement at least one server that uses WebSocket as the transport layer.

A quick survey of the STOMP landscape in Rust shows that there are plenty of STOMP libraries, around 20 on lib.rs at the time of writing. I had a quick look at many of them, but they all have something that make them not a great fit for what I need.

There is a trend here: Almost every library except for wstomp and stomp-parser use TCP transport, and I need to use WebSocket as the transport layer. The crate for WebSocket communication I want to use is tokio-tungstenite but no existing STOMP library I found offers that.

Furthermore, every library except for stomp-parser and stomp are tied to a specific async runtime. At Spiideo we are building more and more cross-platform functionality in Rust that is supposed to work across iOS, Web, Desktop, Backend, and Android. Using Tokio or other async runtimes in the browser is not worth it in my experience; wasm-bindgen-futures is enough if you design your software to be async without using runtime specific functionality.

Of the existing choices, none fit the bill.

It would be great to have a STOMP library that is pluggable. Since STOMP is such a simple protocol (it really is very simple), it should be possible to build a library that is more than a parser, but also doesn't go all the way to mandate a specific transport protocol.

Finding a fork

Out of the available choices, async-stomp seems the best fit for what I want to do. async-stomp was relicensed to use the EUPL license in addition to the original MIT license. I'm very much not a license expert, but the EUPL seems like a reasonable license. I am actually a fan of the EU's work in this field, especially the DMA Enforcement Team, so I'm not even going to try to do anything about the license.

Comically, async-stomp is a fork of tokio-stomp-2, which is a fork of tokio-stomp. I don't think it is out of character to fork it once more.

Fun quote I found in the tokio-stomp-2 repository while researching 👆. At what point am I expected to know what I am doing?

How do we fork?

There are some juicy parts that we are going to need and some parts that we are going to rip right out of async-stomp. The main things to keep are the Frame and Message types, both of which represent a STOMP frame: A Message is the high-level variety and Frame contains the bytes received or to-be-sent. The throwing away part is easy: anything that has to do with TCP transport or Tokio.

The Message type comes in two flavors: Message<FromServer> and Message<ToServer>. The nomenclature is client centric, but I don't think renaming Message<FromServer> to Message<ToClient> would help. Either way you are doing some sort of purpose inversion in your head when implementing either the client or server parts. Adding type aliases might help, but I don't think it would help much.

Designing an agnostic interface

The two axes I want to design along are freedom to choose whatever transport you want and the madness to use whatever async runtime you want.

We are going to need two simple-to-implement traits for the transport layer, one for the client transport and one for the server transport. On top of that we are going to have handles for the client and server parts of STOMP.

The ClientTransport and ServerTransport traits are similar from a design point-of-view so there is no use exploring the design of more than one; for convenience we will look at the client side of things.

The implementation can be found at github.com/bes/stomp-agnostic.

ClientTransport

An important detail for the transport layer is going to be an escape hatch for dealing with protocol-specific communication. For example, a WebSocket connection has Ping and Pong messages, which are not part of STOMP but need to be handled if they come down the wire.

Here's the trait for ClientTransport.

#[async_trait]
pub trait ClientTransport: Send + Sync {
     A side channel to shuffle arbitrary data that is not part of the STOMP communication,
     e.g. WebSocket Ping/Pong.
    type ProtocolSideChannel;

    async fn write(&mut self, message: Message<ToServer>) -> Result<(), WriteError>;
    async fn read(&mut self) -> Result<ReadData<Self::ProtocolSideChannel>, ReadError>;
}

Reading is a matter of getting bytes from some source and then buffering until we have a full Message<FromServer>. Writing similarly is a matter of converting the given Message<ToServer> to bytes using the .into_bytes method and sending them down the wire. Taking a Message<ToServer> instead of just raw bytes for .write makes it impossible (or at least very hard) to misuse the transport trait.

ReadData is a simple enum; either bytes (STOMP data) or some special type T (Self::ProtocolSideChannel in our case) that belongs to the transport protocol.

/// Data coming down the line from the transport layer. When the transport layer is
/// e.g. WebSocket, custom data such as Ping/Pong can be handled separately from STOMP data
/// by using the `Custom` variant.
#[derive(Debug)]
pub enum ReadData<T> {
    Binary(Bytes),
    Custom(T),
}

The ProtocolSideChannel can be as simple as () but for WebSocket it will need to be an enum for the special cases.

enum WebsocketProto {
    Ping(Bytes),
    Pong(Bytes),
}

Notice that to keep the trait simplish, I decided not to be able to send ProtocolSideChannel data through the ClientTransport trait, only to receive it through read. This doesn't stop us from sending a WebSocket Ping though, since the design of the handle allows us to temporarily (or permanently) drop down to transport level and get a hold of the original type.

The magic, if you will, is in transport::client::BufferedTransport

pub(crate) struct BufferedTransport<T>
where
    T: ClientTransport,
    T::ProtocolSideChannel: Debug,
{
    transport: T,
    buffer: BytesMut,
}

BufferedTransport holds on to an implementation of ClientTransport. As we receive data from the transport, it's buffered in a BytesMut until we can read a whole Frame. When we are not waiting on a .read or .write we can get a hold of the underlying transport through BufferedTransport

pub(crate) fn into_transport(self) -> T {
    self.transport
}

pub(crate) fn as_mut_inner(&mut self) -> &mut T {
    &mut self.transport
}

Unfortunately BufferedTransport is crate-private, but the methods are proxied through ClientStompHandle.

ClientStompHandle

The handle type is really just a very thin wrapper around the BufferedTransport, but you can only get a ClientStompHandle using the ClientStompHandle::connect associated function, which makes sure to do a successful STOMP client handshake.

Once you have a ClientStompHandle it's time to rock'n'roll

impl<T> ClientStompHandle<T>
where
    T: ClientTransport,
    T::ProtocolSideChannel: Debug,
{
     Send a STOMP message through the underlying transport
    pub async fn send_message(&mut self, message: Message<ToServer>) -> Result<(), WriteError> {
        self.transport.send(message).await
    }

     Read a STOMP message from the underlying transport
    pub async fn read_response(
        &mut self,
    ) -> Result<ServerResponse<T::ProtocolSideChannel>, ReadError> {
        self.transport.next().await
    }

Implementing a WebSocket client transport

There are simple and functional, but not fully fleshed out, examples in the stomp-agnostic repository. The client example code uses tokio-tungstenite to create a WebSocket client and the server example uses axum to expose an endpoint that can be turned into a WebSocket connection.

Since there are only three server frames to consider, plus the underlying WebSocket protocol, it doesn't take much to implement the full ClientTransport

#[async_trait]
impl ClientTransport for WsTransport {
    type ProtocolSideChannel = WebsocketProto;

    async fn write(
        &mut self,
        message: stomp_agnostic::Message<ToServer>,
    ) -> Result<(), WriteError> {
        let bytes = message.into_bytes();
                        let _ = str::from_utf8(&bytes)?;
        let ws_message = Message::Text(
                        unsafe { Utf8Bytes::from_bytes_unchecked(bytes) },
        );
        self.ws_stream
            .send(ws_message)
            .await
            .map_err(|e| WriteError::Other(anyhow!(e)))
    }

    async fn read(&mut self) -> Result<ReadData<Self::ProtocolSideChannel>, ReadError> {
        loop {
            let message = self
                .ws_stream
                .next()
                .await
                .transpose()
                .map_err(|e| ReadError::Other(anyhow!(e)))?;
            if let Some(message) = message {
                match message {
                    Message::Text(utf8_bytes) => {
                        return Ok(ReadData::Binary(Bytes::copy_from_slice(
                            utf8_bytes.as_bytes(),
                        )));
                    }
                    Message::Binary(bytes) => {
                        return Ok(ReadData::Binary(bytes));
                    }
                    Message::Ping(data) => {
                        return Ok(ReadData::Custom(WebsocketProto::Ping(data)));
                    }
                    Message::Pong(data) => {
                        return Ok(ReadData::Custom(WebsocketProto::Pong(data)));
                    }
                    Message::Close(_) => {
                        return Err(ReadError::ConnectionClosed);
                    }
                    Message::Frame(_) => {
                        return Err(ReadError::UnexpectedMessage);
                    }
                }
            }
        }
    }
}

Fifty-seven lines, including a lot of curly braces.

A note on cancel safety

Rust cancel safety can be tricky, so be sure to understand the details before implementing ClientTransport or ServerTransport. RFD-400 Dealing with cancel safety in async Rust is fantastic and is a very good resource on the subject.

In short, If a function isn't cancel safe there is a risk of leaving the system (your program, your customer's program) in a broken state, a state where some property of the system has been violated.

Let's take a quick look at ClientTransport as an example again:

async fn write(&mut self, message: Message<ToServer>) -> Result<(), WriteError>;
async fn read(&mut self) -> Result<ReadData<Self::ProtocolSideChannel>, ReadError>;

The implementor's responsibility is to make sure that .write and .read are cancel safe. It's a matter of being careful: You are probably going to use some other library (I used tokio-tungstenite and axum in the example code in the repo) to implement the functions. It's your responsibility to figure out their cancel safety properties.

Sending and receiving messages

The best part of stomp-agnostic is how easy it is to use once the transport has been implemented.

I'm hand-waving a bit here, but it doesn't need to be more complicated than this to handle sending and receiving messages as a STOMP-over-WebSocket client

let mut stomp = ClientStompHandle::connect( /* ... */).await?
loop {
    select! {
        _ = cancellation.cancelled() => {
            break;
        },
        message_to_send = internal_message_channel.recv() => {
            match message_to_send {
                Some(message) => stomp.send_message(message).await?,
                None => break,
            }
        }
        response_result = stomp.read_response() => {
            let response = response_result?;
            match response {
                Response::Message(message) => {
                    match message.content {
                        FromServer::Message { body, .. } => {
                            

What now?

Having this library enabled me to replace our very old iOS STOMP client library by writing a few glue functions using UniFFI. All the lovely Rust async cancellation, and a simple library that I am familiar with that can be reused for many other cases.

I'm currently also working on a desktop application for Spiideo that has a STOMP-over-WebSocket server, obviously powered by stomp-agnostic. The magic of loop, select!, CancellationToken, and channels makes it easy to build robust software that is easy to understand. Certainly a far cry from callback based concurrent programming. Rust manages to beat any other programming language I've tried, including async Swift and async TypeScript.

I'm hoping that stomp-agnostic will not only be a tool for me to keep building a variety of Rust STOMP clients and servers, but also for others that have needs that stray from the available STOMP Rust libraries.