# Getting Started (/connectors/stomp/tutorials/getting-started)



Get a message flowing through the KubeMQ STOMP connector in minutes. You point a standard STOMP
client at the broker, `SUBSCRIBE` to a `/topic/` destination, `SEND` a message to a matching
destination, and watch it arrive — all over the native STOMP wire, with no KubeMQ SDK. This
walkthrough takes you from a running server to a verified pub/sub round-trip.

## Prerequisites [#prerequisites]

* A running **kubemq-server** with the STOMP connector **enabled** and reachable on **port 61613** (plain TCP).
  The connector is **opt-in (disabled by default)** — see the enable step below.
* One of the STOMP clients below for your language (the examples pin a native client per
  language — there is no KubeMQ SDK).

## Enable the connector [#enable-the-connector]

The STOMP connector is **disabled by default** — a stock kubemq-server does **not** bind
the STOMP listener until you turn it on. Enable it with its enable variable:

<RunKubeMQ ports="[61613, 61614, 50000]" env="{ CONNECTORS_STOMP_ENABLE: 'true' }" />

<Callout type="warn">
  **The enable variable is `CONNECTORS_STOMP_ENABLE` — spell it verbatim, with the underscore
  between `CONNECTORS` and `STOMP`.** The form `CONNECTORSSTOMP_ENABLE` (without underscore)
  does **not** bind and is silently ignored. For Kubernetes, set `spec.stomp.enabled: true`
  in the `KubemqCluster` CR.
</Callout>

Bring up a throwaway local broker with STOMP enabled:

Every example reads a single environment variable for the broker endpoint. The scheme selects
the transport — `tcp://` (61613) or `tls://` (61614):

```bash
# default: tcp://localhost:61613
export KUBEMQ_STOMP_URL="tcp://localhost:61613"
```

<Callout type="info">
  **Verify the listener is actually up.** The connector loader is *availability-first* — if the
  port fails to bind, the server logs an error and keeps running **without** STOMP rather than
  crashing. Do not infer the listener from a successful server boot: confirm it via the `/stomp`
  dashboard, the `kubemq_stomp_connections` Prometheus gauge, or
  `GET /api/stomp/connections`. See
  [Connections endpoint](/connectors/stomp/reference/connections-endpoint).
</Callout>

To **disable** the STOMP connector after enabling it, set its enable variable to `false`:

<RunKubeMQ variant="disable" env="{ CONNECTORS_STOMP_ENABLE: 'false' }" />

When `Enable` is `false`, no STOMP listener binds and the rest of the STOMP config is skipped.
See [Configuration](/connectors/stomp/concepts/configuration) for the full settings list.

## How it works [#how-it-works]

Every connection opens with one `CONNECT` frame that negotiates the protocol version and
heartbeat; the connector replies with `CONNECTED`. A subscriber registers a `/topic/`
destination; a publisher `SEND`s to a destination with the same prefix. The connector resolves
the prefix (`/topic/`) to the Events pattern, joins the remaining segments with `.` for the
channel, and delivers every message to matching subscribers.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher<br/>SEND /topic/demo&#x22;]
CONN[&#x22;STOMP connector<br/>:61613&#x22;]
CH{{&#x22;Events channel<br/>demo&#x22;}}
SUB[&#x22;Subscriber<br/>SUBSCRIBE /topic/demo&#x22;]

PUB -- &#x22;SEND&#x22; --> CONN
CONN -- &#x22;(events, demo)&#x22; --> CH
CH -. &#x22;fan-out&#x22; .-> CONN
CONN -. &#x22;MESSAGE&#x22; .-> SUB

class PUB,SUB client
class CONN connector
class CH events
`"
/>

*A `SEND` to `/topic/demo` maps to the Events channel `demo`; the connector fans the message out as a `MESSAGE` frame to every subscriber on that destination.*

## Steps [#steps]

<Steps>
  <Step>
    ### Connect to the broker [#connect-to-the-broker]

    Open a STOMP connection to the endpoint in `KUBEMQ_STOMP_URL`. The connector negotiates the
    highest common version of 1.0/1.1/1.2; the examples request `accept-version:1.2`. In the default
    no-auth mode any `login` / `passcode` works (including empty) — `login` is freeform and only
    derives the session id when there are no auth claims.

    The language tabs across all three steps run the **complete** round-trip from a single program:
    connect a subscriber and a publisher, subscribe to `/topic/demo`, send one message, and confirm
    the subscriber receives it.

    <Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
      <Tab value="Go">
        ```go
        package main

        import (
        	"fmt"
        	"log"
        	"os"
        	"time"

        	"github.com/go-stomp/stomp/v3"
        )

        func stompAddr() string {
        	url := os.Getenv("KUBEMQ_STOMP_URL")
        	if url == "" {
        		url = "tcp://localhost:61613"
        	}
        	return url[len("tcp://"):]
        }

        func dial(login string) (*stomp.Conn, error) {
        	return stomp.Dial("tcp", stompAddr(),
        		stomp.ConnOpt.AcceptVersion(stomp.V12),
        		stomp.ConnOpt.Login(login, ""))
        }

        func main() {
        	const destination = "/topic/demo" // Events pattern, channel "demo"

        	// SUBSCRIBER — connect and subscribe before publishing (Events is at-most-once).
        	sub, err := dial("go-sub")
        	if err != nil {
        		log.Fatalf("sub connect: %v", err)
        	}
        	subscription, err := sub.Subscribe(destination, stomp.AckAuto)
        	if err != nil {
        		log.Fatalf("subscribe: %v", err)
        	}
        	fmt.Printf("[sub] subscribed to %s\n", destination)
        	time.Sleep(300 * time.Millisecond) // let the subscription register

        	// PUBLISHER — connect and SEND one message.
        	pub, err := dial("go-pub")
        	if err != nil {
        		log.Fatalf("pub connect: %v", err)
        	}
        	if err := pub.Send(destination, "application/json",
        		[]byte(`{"message":"hello"}`), stomp.SendOpt.Receipt); err != nil {
        		log.Fatalf("send: %v", err)
        	}
        	fmt.Printf("[pub] sent to %s\n", destination)

        	// RECEIVE — wait for the MESSAGE frame.
        	select {
        	case msg := <-subscription.C:
        		fmt.Printf("[sub] received: %s (destination=%s)\n", string(msg.Body), msg.Destination)
        	case <-time.After(10 * time.Second):
        		log.Fatal("timed out waiting for the event")
        	}

        	_ = pub.Disconnect()
        	_ = sub.Disconnect()
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python
        import os
        import queue
        import time

        import stomp


        def endpoint() -> tuple[str, int]:
            url = os.environ.get("KUBEMQ_STOMP_URL", "tcp://localhost:61613")
            host, port = url.split("://", 1)[1].split(":", 1)
            return host, int(port)


        class Listener(stomp.ConnectionListener):
            def __init__(self) -> None:
                self.messages: queue.Queue = queue.Queue()

            def on_message(self, frame) -> None:
                self.messages.put(frame)


        def connect(login: str, listener: stomp.ConnectionListener | None = None) -> stomp.Connection:
            conn = stomp.Connection([endpoint()], heartbeats=(10000, 10000))
            if listener is not None:
                conn.set_listener("", listener)
            conn.connect(login=login, passcode="", wait=True)
            return conn


        def main() -> None:
            destination = "/topic/demo"  # Events pattern, channel "demo"

            # SUBSCRIBER — connect and subscribe first.
            listener = Listener()
            sub = connect("py-sub", listener)
            sub.subscribe(destination=destination, id="sub-1", ack="auto")
            print(f"[sub] subscribed to {destination}")
            time.sleep(0.3)

            # PUBLISHER — connect and SEND.
            pub = connect("py-pub")
            pub.send(destination=destination, body='{"message":"hello"}',
                     content_type="application/json")
            print(f"[pub] sent to {destination}")

            frame = listener.messages.get(timeout=10)
            body = frame.body if isinstance(frame.body, str) else frame.body.decode()
            print(f"[sub] received: {body} (destination={frame.headers['destination']})")

            pub.disconnect()
            sub.disconnect()


        if __name__ == "__main__":
            main()
        ```
      </Tab>

      <Tab value="Java">
        ```java
        import java.net.URI;
        import java.util.concurrent.LinkedBlockingQueue;
        import java.util.concurrent.TimeUnit;
        import java.lang.reflect.Type;

        import org.springframework.messaging.simp.stomp.StompHeaders;
        import org.springframework.messaging.simp.stomp.StompSession;
        import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
        import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient;

        public final class Main {
            public static void main(String[] args) throws Exception {
                String url = System.getenv().getOrDefault("KUBEMQ_STOMP_URL", "tcp://localhost:61613");
                URI uri = URI.create(url);
                String destination = "/topic/demo"; // Events pattern, channel "demo"

                ReactorNettyTcpStompClient client =
                        new ReactorNettyTcpStompClient(uri.getHost(), uri.getPort());
                LinkedBlockingQueue<String> inbox = new LinkedBlockingQueue<>();

                // SUBSCRIBER
                StompSession sub = client.connect(new StompSessionHandlerAdapter() {}).get();
                sub.subscribe(destination, new StompSessionHandlerAdapter() {
                    @Override public Type getPayloadType(StompHeaders headers) { return byte[].class; }
                    @Override public void handleFrame(StompHeaders headers, Object payload) {
                        inbox.add(new String((byte[]) payload));
                    }
                });
                System.out.printf("[sub] subscribed to %s%n", destination);
                Thread.sleep(300);

                // PUBLISHER
                StompSession pub = client.connect(new StompSessionHandlerAdapter() {}).get();
                StompHeaders headers = new StompHeaders();
                headers.setDestination(destination);
                headers.add("content-type", "application/json");
                pub.send(headers, "{\"message\":\"hello\"}".getBytes());
                System.out.printf("[pub] sent to %s%n", destination);

                String body = inbox.poll(10, TimeUnit.SECONDS);
                System.out.printf("[sub] received: %s%n", body);

                pub.disconnect();
                sub.disconnect();
                client.shutdown();
            }
        }
        ```
      </Tab>

      <Tab value="JavaScript">
        ```typescript
        import { connect, type Client } from "stompit";

        function endpoint(): { host: string; port: number } {
          const url = new URL(process.env["KUBEMQ_STOMP_URL"] ?? "tcp://localhost:61613");
          return { host: url.hostname, port: Number(url.port) || 61613 };
        }

        function dial(login: string): Promise<Client> {
          const { host, port } = endpoint();
          return new Promise((resolve, reject) => {
            connect({ host, port, connectHeaders: { "accept-version": "1.2",
              "heart-beat": "10000,10000", login } },
              (err, c) => (err ? reject(err) : resolve(c)));
          });
        }

        async function main(): Promise<void> {
          const destination = "/topic/demo"; // Events pattern, channel "demo"

          // SUBSCRIBER
          const sub = await dial("js-sub");
          const received = new Promise<string>((resolve, reject) => {
            sub.subscribe({ destination, ack: "auto" }, (err, message) => {
              if (err) return reject(err);
              message.readString("utf-8", (e, body) => (e ? reject(e) : resolve(body ?? "")));
            });
          });
          console.log(`[sub] subscribed to ${destination}`);
          await new Promise((r) => setTimeout(r, 300));

          // PUBLISHER
          const pub = await dial("js-pub");
          const frame = pub.send({ destination, "content-type": "application/json" });
          frame.write('{"message":"hello"}');
          frame.end();
          console.log(`[pub] sent to ${destination}`);

          console.log(`[sub] received: ${await received}`);
          pub.disconnect();
          sub.disconnect();
        }

        main().catch((err) => {
          console.error(err);
          process.exit(1);
        });
        ```
      </Tab>

      <Tab value="C#">
        ```csharp
        using System.Text;
        using Apache.NMS;
        using Stomp.Net;

        static string BrokerUri()
        {
            var url = Environment.GetEnvironmentVariable("KUBEMQ_STOMP_URL") ?? "tcp://localhost:61613";
            var u = new Uri(url);
            var transport = u.Scheme == "tls" ? "ssl" : "tcp";
            return $"{transport}://{u.Host}:{(u.Port > 0 ? u.Port : 61613)}";
        }

        const string destination = "/topic/demo"; // Events pattern, channel "demo"

        var factory = new ConnectionFactory(BrokerUri(), new StompConnectionSettings());
        var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);

        // SUBSCRIBER
        using var subConn = factory.CreateConnection();
        subConn.Start();
        using var subSession = subConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
        using var consumer = subSession.CreateConsumer(subSession.GetTopic(destination));
        consumer.Listener += msg =>
            received.TrySetResult(Encoding.UTF8.GetString(((IBytesMessage)msg).Content));
        Console.WriteLine($"[sub] subscribed to {destination}");
        await Task.Delay(300);

        // PUBLISHER
        using var pubConn = factory.CreateConnection();
        pubConn.Start();
        using var pubSession = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
        using var producer = pubSession.CreateProducer(pubSession.GetTopic(destination));
        var message = pubSession.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"message\":\"hello\"}"));
        message.StompType = "application/json";
        producer.Send(message);
        Console.WriteLine($"[pub] sent to {destination}");

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
        Console.WriteLine($"[sub] received: {await received.Task.WaitAsync(cts.Token)}");
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby
        require "stomp"
        require "uri"
        require "timeout"

        uri = URI.parse(ENV.fetch("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
        destination = "/topic/demo" # Events pattern, channel "demo"

        def connect(uri, login)
          Stomp::Client.new(
            hosts: [{ host: uri.host, port: uri.port }],
            connect_headers: { "accept-version" => "1.2", "heart-beat" => "10000,10000",
                               "login" => login, "passcode" => "" },
          )
        end

        inbox = Queue.new

        # SUBSCRIBER
        sub = connect(uri, "rb-sub")
        sub.subscribe(destination, id: "sub-1", ack: "auto") { |msg| inbox << msg.body }
        puts "[sub] subscribed to #{destination}"
        sleep 0.3

        # PUBLISHER
        pub = connect(uri, "rb-pub")
        pub.publish(destination, '{"message":"hello"}', { "content-type" => "application/json" })
        puts "[pub] sent to #{destination}"

        body = Timeout.timeout(10) { inbox.pop }
        puts "[sub] received: #{body}"

        pub.close
        sub.close
        ```
      </Tab>

      <Tab value="Rust">
        ```rust
        use async_stomp::client::Connector;
        use async_stomp::{FromServer, ToServer};
        use futures::{SinkExt, StreamExt};
        use std::time::Duration;
        use tokio::time::timeout;

        async fn dial(host_port: &str, login: &str) -> Result<
            impl SinkExt<async_stomp::Message<ToServer>> + StreamExt, Box<dyn std::error::Error>> {
            Ok(Connector::builder()
                .server(host_port)
                .login(login.to_string())
                .passcode(String::new())
                .connect()
                .await?)
        }

        #[tokio::main]
        async fn main() -> Result<(), Box<dyn std::error::Error>> {
            let url = std::env::var("KUBEMQ_STOMP_URL").unwrap_or_else(|_| "tcp://localhost:61613".into());
            let host_port = url.split("://").nth(1).unwrap_or("localhost:61613").to_string();
            let destination = "/topic/demo"; // Events pattern, channel "demo"

            // SUBSCRIBER
            let mut sub = Connector::builder().server(&host_port)
                .login("rust-sub".into()).passcode(String::new()).connect().await?;
            sub.send(ToServer::Subscribe {
                destination: destination.to_string(),
                id: "sub-1".to_string(),
                ack: None,
            }).await?;
            println!("[sub] subscribed to {destination}");
            tokio::time::sleep(Duration::from_millis(300)).await;

            // PUBLISHER
            let mut pubc = Connector::builder().server(&host_port)
                .login("rust-pub".into()).passcode(String::new()).connect().await?;
            pubc.send(ToServer::Send {
                destination: destination.to_string(),
                transaction: None,
                headers: Some(vec![("content-type".into(), "application/json".into())]),
                body: Some(br#"{"message":"hello"}"#.to_vec()),
            }).await?;
            println!("[pub] sent to {destination}");

            // RECEIVE
            if let Ok(Some(Ok(msg))) = timeout(Duration::from_secs(10), sub.next()).await {
                if let FromServer::Message { body, .. } = msg.content {
                    println!("[sub] received: {}", String::from_utf8_lossy(&body.unwrap_or_default()));
                }
            }

            pubc.send(ToServer::Disconnect { receipt: None }).await?;
            Ok(())
        }
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Send a message [#send-a-message]

    The publisher in the program above `SEND`s one message to `/topic/demo`. The prefix `/topic/`
    selects the Events pattern, and the remaining segments become the KubeMQ channel with `/`
    translated to `.` — so `/topic/demo` lands on channel `demo`. Attaching a `receipt:` header
    makes the connector return a `RECEIPT` once it has **accepted** the frame (this confirms
    acceptance, not consumer delivery).
  </Step>

  <Step>
    ### Subscribe and verify [#subscribe-and-verify]

    The subscriber registers the destination `/topic/demo`. When the message arrives it is
    delivered as a `MESSAGE` frame whose `destination` header is canonicalized to the primary form,
    and the program prints it and exits:

    ```text
    [sub] subscribed to /topic/demo
    [pub] sent to /topic/demo
    [sub] received: {"message":"hello"} (destination=/topic/demo)
    ```

    Events is fire-and-forget pub/sub: subscribe **before** you publish, or the message is gone. For
    persistence and replay-on-reconnect, use the Events-Store pattern (`/topic-store/`) instead.
  </Step>
</Steps>

<Callout type="info">
  Wildcard subscriptions are accepted on the **Events** pattern only, are **subscribe-only**, and
  use the message broker's native wildcard syntax (`*` = one segment, `>` = the final tail) —
  there is no MQTT-style `+`/`#`. Avoid a literal `.` in a destination segment: `/topic/a.b` and
  `/topic/a/b` both map to channel `a.b`, and egress always emits the slash form. See
  [Destination mapping](/connectors/stomp/how-to/destination-mapping).
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Configuration" href="/connectors/stomp/concepts/configuration" description="The 11 connector settings, the CONNECTORS_STOMP_ENABLE spelling, ports, and validation." />

  <Card title="Architecture" href="/connectors/stomp/concepts/architecture" description="The embedded STOMP server, the frame codec, version negotiation, and the destination router." />

  <Card title="Events" href="/connectors/stomp/how-to/events" description="Fire-and-forget pub/sub over STOMP — fan-out, wildcards, and the slash→dot mapping." />

  <Card title="Queues" href="/connectors/stomp/how-to/queues" description="Competing-consumer queues over STOMP with ack modes, receipts, and redelivery." />
</Cards>