Migrating from STOMP

Migrate a STOMP client application onto KubeMQ's native STOMP connector. The connector implements STOMP 1.0 / 1.1 / 1.2 over plain TCP and TLS — no broker middleware is interposed, so your STOMP client connects directly to KubeMQ. This is a drop-in, endpoint-only migration: change the host and port, keep the same STOMP client library and application code.

The guide is broker-agnostic — it applies to any STOMP client driving any STOMP broker today: stomp.py, @stomp/stompjs, go-stomp/stomp/v3, the Ruby stomp gem, Stomp.Net, or a Spring broker-relay front-end. Only the library API differs; the wire protocol and the migration steps are identical.

Overview

The KubeMQ STOMP connector is a faithful STOMP 1.0 / 1.1 / 1.2 raw-TCP endpoint. The code examples on this page target stomp.py 8.x (Python) as the canonical client; other clients connect identically.

The connector is opt-in — Connectors.Stomp.Enable defaults to false, so a stock kubemq-server does not bind the STOMP listener until you turn it on:

docker run -d \  --name kubemq \  -p 61613:61613 \  -p 61614:61614 \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  -e CONNECTORS_STOMP_ENABLE=true \  europe-docker.pkg.dev/kubemq/images/kubemq:next

For the full wire-protocol contract (heartbeats, receipts, header/tag mapping, ack token internals, error table, metrics catalog) see Capabilities, Destination grammar, and Error frames. For configuration fields and environment variables see the configuration reference.

Compatibility Matrix

This is the STOMP column of the cross-protocol compatibility matrix in the Migration hub. It is self-contained — read it as the at-a-glance answer to "what migrates?"

¹ No client-settable DLQ over STOMP. The STOMP connector never sets MaxReceiveQueue on published messages, so a poison message that exceeds MaxReceiveCount is silently dropped by the broker — it is not delivered to any consumable dead-letter address. If you need client-facing dead-letter behaviour, use the RabbitMQ (DLX) or AWS (redrive) path instead.

Connection / Endpoint Migration

The only change required is the host and port. The STOMP protocol remains identical.

stomp.py 8.x — before

import stomp

conn = stomp.Connection([("stomp-host", 61613)])
conn.connect("user", "password", wait=True)

stomp.py 8.x — after (KubeMQ)

import stomp

conn = stomp.Connection([("kubemq-host", 61613)])
conn.connect("user", jwt_token, wait=True)   # jwt_token ignored when auth disabled

For TLS, use stomp.Connection([("kubemq-host", 61614)], use_ssl=True).

Concept & Destination Mapping

The STOMP destination header determines both the KubeMQ messaging pattern and the KubeMQ channel name. Normalization strips one leading /, splits on /, joins remaining segments with ..

Prefix table

A destination whose first segment does not match any prefix maps to Connectors.Stomp.DefaultPattern (default events). Setting DefaultPattern = none rejects those destinations with ERROR invalid destination.

Destination examples

Wildcards

The broker's native wildcards — * (single segment) and > (tail) — are supported on SUBSCRIBE, Events pattern only (/topic/* or /topic/orders.>). These are the STOMP destination grammar's own wildcards; there is no MQTT-style + / #. Wildcards on SEND, on any other pattern, or a misplaced > are rejected with ERROR invalid destination.

Durable subscriptions → Events Store + replay headers

STOMP has no built-in durable-subscription mechanism. In KubeMQ the equivalent is subscribing to a /topic-store/ destination with a start-from replay header:

Example: subscribe from the beginning of an Events Store channel:

conn.subscribe(
    destination="/topic-store/audit-log",
    id="sub-1",
    ack="client-individual",
    headers={"start-from": "first"},
)

Ack modes

Set per subscription via the ack header on SUBSCRIBE.

ACK frames on Events or Events Store subscriptions are accepted as a no-op (those patterns do not track per-message delivery state).

Canonical Client Example (stomp.py 8.x)

This example uses the following stomp.py 8.x API symbols: stomp.Connection, conn.set_listener, conn.connect, conn.send, conn.subscribe, conn.ack, conn.disconnect, stomp.ConnectionListener.on_message.

Publish to a queue

import stomp

KUBEMQ_HOST = "kubemq-host"
KUBEMQ_PORT = 61613
JWT_TOKEN = "..."            # omit or use any string when auth is disabled

conn = stomp.Connection([(KUBEMQ_HOST, KUBEMQ_PORT)])
conn.connect("myapp", JWT_TOKEN, wait=True)

conn.send(
    destination="/queue/orders",
    body="order payload",
    headers={"content-type": "text/plain"},
)

conn.disconnect()

Subscribe and consume from a queue (client-individual ack)

import stomp

KUBEMQ_HOST = "kubemq-host"
KUBEMQ_PORT = 61613
JWT_TOKEN = "..."

class QueueListener(stomp.ConnectionListener):
    def __init__(self, conn):
        self._conn = conn

    def on_message(self, frame):
        print("received:", frame.body)
        # Acknowledge the individual message
        self._conn.ack(frame.headers["ack"])

    def on_error(self, frame):
        print("error:", frame.headers.get("message"))

conn = stomp.Connection([(KUBEMQ_HOST, KUBEMQ_PORT)])
conn.set_listener("", QueueListener(conn))
conn.connect("myapp", JWT_TOKEN, wait=True)

conn.subscribe(
    destination="/queue/orders",
    id="sub-orders",
    ack="client-individual",
)

input("Press Enter to stop...\n")
conn.disconnect()

Pub/sub via Events

import stomp, threading

KUBEMQ_HOST = "kubemq-host"
KUBEMQ_PORT = 61613

received = threading.Event()

class EventListener(stomp.ConnectionListener):
    def on_message(self, frame):
        print("event:", frame.body)
        received.set()

conn = stomp.Connection([(KUBEMQ_HOST, KUBEMQ_PORT)])
conn.set_listener("", EventListener())
conn.connect(wait=True)

conn.subscribe(destination="/topic/orders.created", id="sub-1", ack="auto")
conn.send(destination="/topic/orders.created", body="event payload")

received.wait(timeout=5)
conn.disconnect()

RPC (Commands) with reply-to

The reply subscription must be created before sending the RPC request. See What Does NOT Migrate / Deviations for details.

import stomp, threading, uuid

KUBEMQ_HOST = "kubemq-host"
KUBEMQ_PORT = 61613
JWT_TOKEN = "..."

reply_event = threading.Event()
reply_body = None

class RpcListener(stomp.ConnectionListener):
    def on_message(self, frame):
        global reply_body
        reply_body = frame.body
        reply_event.set()

reply_dest = f"/reply/{uuid.uuid4()}"

conn = stomp.Connection([(KUBEMQ_HOST, KUBEMQ_PORT)])
conn.set_listener("", RpcListener())
conn.connect("myapp", JWT_TOKEN, wait=True)

# Step 1: subscribe to the reply destination BEFORE sending the request
conn.subscribe(destination=reply_dest, id="reply-sub", ack="auto")

# Step 2: send the RPC request with reply-to and correlation-id headers
conn.send(
    destination="/command/payments/process",
    body="payment request",
    headers={
        "reply-to": reply_dest,
        "correlation-id": str(uuid.uuid4()),
        "timeout": "5000",       # ms; capped at RpcTimeoutSeconds * 1000
    },
)

reply_event.wait(timeout=10)
print("reply:", reply_body)
conn.disconnect()

Security

Authentication (connect-time only). When Authentication.Enable = true, the CONNECT frame passcode header must carry a valid KubeMQ JWT. The login header is recorded for audit purposes. Auth failure results in ERROR authentication failed + connection close. When authentication is disabled, any credentials are accepted — no auth is performed.

Token expiry does not terminate an established connection (connect-time-only auth, consistent with the AMQP and MQTT connectors).

Authorization. When Authorization.Enable = true, SEND enforces the Write permission and SUBSCRIBE enforces Read on ControlRecord{Resource: <pattern>, ClientID, Channel}. Casbin policies must cover stomp-* client IDs for the mapped channels. Reply (/reply/...) destinations are authorization-exempt (connection-local).

No-auth exposure note. When Authentication.Enable = false (the server default), the STOMP listener accepts any login / passcode. If the server is reachable from untrusted networks, enable authentication or restrict access with a firewall.

TLS. Port 61614 is active only when Connectors.Stomp.TlsPort is set and Security is configured (mode ≠ none). Plain TCP on 61613 carries no transport encryption.

Enabling the connector.

[Connectors.Stomp]
  Enable = true
  Port   = "61613"
  TlsPort = "61614"

For the full auth and Casbin authorization setup, see Authentication & Security.

What Does NOT Migrate / Deviations

Hard failures (client will receive ERROR + connection close)

Behavioural deviations

Verification Smoke Test

The recipe below uses the stomp.py 8.x snippets from Canonical Client Example and confirms that a message published to a queue arrives at a subscriber.

Steps

1. Start KubeMQ with the STOMP connector enabled:

   docker run -d \  --name kubemq \  -p 61613:61613 \  -p 61614:61614 \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  -e CONNECTORS_STOMP_ENABLE=true \  europe-docker.pkg.dev/kubemq/images/kubemq:next

2. In a terminal, run the consumer script (subscribes to /queue/smoke-test):

   # consumer.py — stomp.py 8.x
   import stomp, time
   
   class Listener(stomp.ConnectionListener):
       def __init__(self, conn):
           self._conn = conn
   
       def on_message(self, frame):
           print("RECEIVED:", frame.body)
           self._conn.ack(frame.headers["ack"])
   
   conn = stomp.Connection([("localhost", 61613)])
   conn.set_listener("", Listener(conn))
   conn.connect(wait=True)
   conn.subscribe(destination="/queue/smoke-test", id="s1", ack="client-individual")
   
   time.sleep(10)    # wait for a message
   conn.disconnect()

3. In a second terminal, publish one message:

   # producer.py — stomp.py 8.x
   import stomp
   
   conn = stomp.Connection([("localhost", 61613)])
   conn.connect(wait=True)
   conn.send(destination="/queue/smoke-test", body="hello from stomp")
   conn.disconnect()
   print("sent")

4. Confirm the consumer terminal prints RECEIVED: hello from stomp.

5. To verify Events Store replay, repeat with /topic-store/smoke-test and headers={"start-from": "first"} on the subscriber. The message should be delivered even if the subscriber connects after the publisher disconnects.

See Also