# Queues (/connectors/stomp/how-to/queues)



Queues are durable, **competing-consumer** work queues over the STOMP connector. SEND to a destination prefixed with `/queue/<channel>` and the connector routes the message to the KubeMQ **Queues** pattern; SUBSCRIBE to the same destination and each message is delivered to **exactly one** consumer (round-robin across the subscribers), acknowledged, and **at-least-once**.

## Overview [#overview]

The `/queue/` prefix selects the Queues pattern; `/queues/` is an accepted alias that egress always canonicalizes back to the primary `/queue/...` form — **lead with `/queue/` in your code**. The remaining destination segments are slash-to-dot joined into the KubeMQ channel: `/queue/jobs/email` → channel `jobs.email`.

A producer SENDs a message; a consumer SUBSCRIBEs with an `ack` header and ACKs each delivery. Many consumers can subscribe to the same `/queue/...` channel: the connector hands each message to **one** of them (round-robin competing-consumer move semantics — not fan-out). Add consumers to scale throughput; each message is still processed once.

| Operation    | STOMP action                               | KubeMQ mapping                        |
| ------------ | ------------------------------------------ | ------------------------------------- |
| Produce      | `SEND /queue/<ch>` (optional `receipt`)    | `SendQueueMessage`                    |
| Consume      | `SUBSCRIBE /queue/<ch>` with an `ack` mode | Credit-driven `Get` poll, round-robin |
| Acknowledge  | `ACK id:<token>`                           | `AckRange` — message removed          |
| Negative-ack | `NACK id:<token>`                          | `NAckRange` — requeued to the tail    |

The RECEIPT on a SEND fires when KubeMQ **accepts** the message — not when a consumer receives it.

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

A producer enqueues messages; competing consumers subscribe with an ack mode, receive a message each (round-robin), do the work, and ACK — the message is removed from the queue. An un-ACKed message is requeued (after the ack-timeout, or immediately on disconnect) and redelivered, never lost.

<Mermaid
  chart="`
graph LR
PROD[&#x22;Producer<br/>(SEND /queue/work)&#x22;]
CONN[&#x22;STOMP connector<br/>:61613&#x22;]
BROKER[&#x22;Message Broker&#x22;]
Q{{&#x22;Queue channel<br/>work&#x22;}}
W1[&#x22;Worker A&#x22;]
W2[&#x22;Worker B&#x22;]

PROD -- &#x22;SEND&#x22; --> CONN
CONN -- &#x22;SendQueueMessage&#x22; --> BROKER
BROKER --> Q
Q -- &#x22;MESSAGE (one, round-robin)&#x22; --> W1
Q -. &#x22;next MESSAGE&#x22; .-> W2
W1 -. &#x22;ACK → AckRange&#x22; .-> CONN

class PROD,W1,W2 client
class CONN connector
class BROKER broker
class Q queue
`"
/>

*Each queued message is delivered to exactly one competing consumer; an `ACK` removes the message, a `NACK` (or ack-timeout) requeues it to the tail.*

## The three ack modes [#the-three-ack-modes]

A `/queue/...` SUBSCRIBE carries an `ack` header; absent → `auto`. The connector accepts exactly three values; anything else → `ERROR "unknown ack mode"` and the connection closes.

| `ack` mode                        | Client action                                                 | KubeMQ downstream                                                                                                          |
| --------------------------------- | ------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- |
| `auto&#x60; &#x2A;(default)*      | none                                                          | the connector immediately acks each delivery; a NACK/requeue happens only if the output buffer is full — **never dropped** |
| `client-individual`               | ACK/NACK **one** message by its token                         | one `AckRange`/`NAckRange` for that delivery                                                                               |
| `client&#x60; &#x2A;(cumulative)* | ACK/NACK that message **and all earlier** on the subscription | grouped by transaction, one `AckRange`/`NAckRange` per transaction                                                         |

**`client-individual` is the recommended reliable default** — every delivery is ACKed (or NACKed) on its own, the simplest at-least-once mental model.

The ACK/NACK token source is version-dependent: STOMP 1.2 uses the opaque `ack` header (a UUID distinct from `message-id`); 1.1 uses `message-id` + `subscription`; 1.0 uses `message-id`. The examples below use 1.2.

## Produce and consume [#produce-and-consume]

Each example SENDs a batch of jobs to a `/queue/...` destination, then subscribes with `ack:client-individual` and ACKs each delivery by its `ack` token. On the happy path every message is ACKed, so none are redelivered. Every client reads the connector endpoint from `KUBEMQ_STOMP_URL` (default `tcp://localhost:61613`).

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

    import (
    	"fmt"
    	"log"
    	"net/url"
    	"os"
    	"time"

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

    const destination = "/queue/jobs/email" // Queues pattern → channel jobs.email
    const count = 5

    func addr() (network, host string) {
    	u, _ := url.Parse(os.Getenv("KUBEMQ_STOMP_URL"))
    	if u == nil || u.Host == "" {
    		return "tcp", "localhost:61613"
    	}
    	return "tcp", u.Host
    }

    func main() {
    	network, host := addr()

    	// 1. Produce — SEND `count` jobs to the queue.
    	pub, err := stomp.Dial(network, host)
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	for i := 1; i <= count; i++ {
    		body := fmt.Sprintf("job-%d", i)
    		if err := pub.Send(destination, "text/plain", []byte(body)); err != nil {
    			log.Fatalf("send: %v", err)
    		}
    	}
    	_ = pub.Disconnect()

    	// 2. Consume — SUBSCRIBE ack:client-individual; ACK each by its 1.2 token.
    	sub, err := stomp.Dial(network, host)
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	defer func() { _ = sub.Disconnect() }()

    	subscription, err := sub.Subscribe(destination, stomp.AckClientIndividual)
    	if err != nil {
    		log.Fatalf("subscribe: %v", err)
    	}

    	for got := 0; got < count; {
    		select {
    		case msg := <-subscription.C:
    			if msg.Err != nil {
    				log.Fatalf("receive: %v", msg.Err)
    			}
    			if msg.Header.Get("redelivered") == "true" {
    				log.Fatalf("unexpected redelivery of %q", string(msg.Body))
    			}
    			if err := sub.Ack(msg); err != nil { // ACK by the 1.2 ack token
    				log.Fatalf("ack: %v", err)
    			}
    			got++
    		case <-time.After(10 * time.Second):
    			log.Fatal("timed out draining the queue")
    		}
    	}
    	fmt.Printf("drained and acked %d jobs\n", count)
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os
    import queue
    from urllib.parse import urlparse

    import stomp

    DESTINATION = "/queue/jobs/email"  # Queues pattern → channel jobs.email
    COUNT = 5


    def endpoint() -> tuple[str, int]:
        parsed = urlparse(os.environ.get("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
        return parsed.hostname or "localhost", parsed.port or 61613


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

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


    def main() -> None:
        host, port = endpoint()

        # 1. Produce — SEND `count` jobs to the queue.
        pub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
        pub.connect(wait=True)
        for i in range(1, COUNT + 1):
            pub.send(DESTINATION, f"job-{i}", content_type="text/plain")
        pub.disconnect()

        # 2. Consume — SUBSCRIBE ack:client-individual; ACK each by its 1.2 token.
        consumer = Consumer()
        sub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
        sub.set_listener("c", consumer)
        sub.connect(wait=True)
        sub.subscribe(DESTINATION, id="jobs", ack="client-individual")

        for _ in range(COUNT):
            frame = consumer.inbox.get(timeout=10)
            if frame.headers.get("redelivered") == "true":
                raise SystemExit(f"unexpected redelivery: {frame.body!r}")
            sub.ack(frame.headers["ack"])  # ACK by the 1.2 ack token
        sub.disconnect()
        print(f"drained and acked {COUNT} jobs")


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

  <Tab value="Java">
    ```java
    import java.lang.reflect.Type;
    import java.nio.charset.StandardCharsets;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;

    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.tcp.reactor.ReactorNettyTcpClient;
    import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient;

    public final class Main {
        private static final String DESTINATION = "/queue/jobs/email"; // → channel jobs.email
        private static final int COUNT = 5;

        public static void main(String[] args) throws Exception {
            String url = System.getenv().getOrDefault("KUBEMQ_STOMP_URL", "tcp://localhost:61613");
            java.net.URI u = java.net.URI.create(url);
            ReactorNettyTcpStompClient client = new ReactorNettyTcpStompClient(u.getHost(),
                    u.getPort() > 0 ? u.getPort() : 61613);

            // 1. Produce — SEND `count` jobs to the queue.
            StompSession pub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
            for (int i = 1; i <= COUNT; i++) {
                StompHeaders h = new StompHeaders();
                h.setDestination(DESTINATION);
                h.add("content-type", "text/plain");
                pub.send(h, ("job-" + i).getBytes(StandardCharsets.UTF_8));
            }
            Thread.sleep(500);
            pub.disconnect();

            // 2. Consume — SUBSCRIBE ack:client-individual; ACK each delivery.
            BlockingQueue<StompHeaders> inbox = new ArrayBlockingQueue<>(COUNT);
            StompSession sub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
            StompHeaders subHeaders = new StompHeaders();
            subHeaders.setDestination(DESTINATION);
            subHeaders.setId("jobs");
            subHeaders.setAck("client-individual");
            sub.subscribe(subHeaders, new StompSessionHandlerAdapter() {
                @Override public Type getPayloadType(StompHeaders headers) { return byte[].class; }
                @Override public void handleFrame(StompHeaders headers, Object payload) { inbox.add(headers); }
            });

            for (int i = 0; i < COUNT; i++) {
                StompHeaders headers = inbox.poll(10, TimeUnit.SECONDS);
                if (headers == null) throw new IllegalStateException("timed out draining the queue");
                if ("true".equals(headers.getFirst("redelivered"))) {
                    throw new IllegalStateException("unexpected redelivery");
                }
                sub.acknowledge(headers.getAck(), true); // ACK by the 1.2 ack token
            }
            sub.disconnect();
            client.stop();
            System.out.printf("drained and acked %d jobs%n", COUNT);
        }
    }
    ```
  </Tab>

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

    const DESTINATION = "/queue/jobs/email"; // Queues pattern → channel jobs.email
    const COUNT = 5;

    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 open(): Promise<Client> {
      const { host, port } = endpoint();
      return new Promise((resolve, reject) => {
        connect({ host, port, connectHeaders: { "accept-version": "1.2", "heart-beat": "10000,10000" } },
          (err, client) => (err ? reject(err) : resolve(client)));
      });
    }

    async function main(): Promise<void> {
      // 1. Produce — SEND `count` jobs to the queue.
      const pub = await open();
      for (let i = 1; i <= COUNT; i++) {
        const frame = pub.send({ destination: DESTINATION, "content-type": "text/plain" });
        frame.write(`job-${i}`);
        frame.end();
      }
      await new Promise<void>((r) => pub.disconnect(() => r()));

      // 2. Consume — SUBSCRIBE ack:client-individual; ACK each delivery.
      const sub = await open();
      let got = 0;
      await new Promise<void>((resolve, reject) => {
        const timer = setTimeout(() => reject(new Error("timed out draining the queue")), 15_000);
        sub.subscribe({ destination: DESTINATION, ack: "client-individual" }, (err, message) => {
          if (err) return reject(err);
          if (message.headers["redelivered"] === "true") return reject(new Error("unexpected redelivery"));
          message.readString("utf-8", (readErr) => {
            if (readErr) return reject(readErr);
            sub.ack(message); // ACK by the 1.2 ack token (stompit tracks it on the frame)
            if (++got === COUNT) {
              clearTimeout(timer);
              resolve();
            }
          });
        });
      });
      await new Promise<void>((r) => sub.disconnect(() => r()));
      console.log(`drained and acked ${COUNT} jobs`);
    }

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

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

    var url = Environment.GetEnvironmentVariable("KUBEMQ_STOMP_URL") ?? "tcp://localhost:61613";
    var uri = new Uri(url);
    const string destination = "/queue/jobs/email"; // Queues pattern → channel jobs.email
    const int count = 5;

    string brokerUri = $"stomp:tcp://{uri.Host}:{(uri.Port > 0 ? uri.Port : 61613)}";
    var factory = new ConnectionFactory(brokerUri) { UserName = "kubemq", Password = "" };

    // 1. Produce — SEND `count` jobs to the queue.
    using (var pubConn = factory.CreateConnection())
    {
        pubConn.Start();
        using var session = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
        using var producer = session.CreateProducer(session.GetQueue(destination));
        for (var i = 1; i <= count; i++)
        {
            var msg = producer.CreateBytesMessage(Encoding.UTF8.GetBytes($"job-{i}"));
            msg.StompType = "text/plain";
            producer.Send(msg);
        }
    }

    // 2. Consume — SUBSCRIBE client-individual; Acknowledge() each delivery.
    using var conn = factory.CreateConnection();
    conn.Start();
    using var consumeSession = conn.CreateSession(AcknowledgementMode.IndividualAcknowledge);
    using var consumer = consumeSession.CreateConsumer(consumeSession.GetQueue(destination));
    for (var i = 0; i < count; i++)
    {
        var msg = consumer.Receive(TimeSpan.FromSeconds(10))
            ?? throw new InvalidOperationException("timed out draining the queue");
        if (msg.Headers.GetValue("redelivered") == "true")
            throw new InvalidOperationException("unexpected redelivery");
        msg.Acknowledge(); // ACK by the 1.2 ack token
    }
    Console.WriteLine($"drained and acked {count} jobs");
    ```
  </Tab>

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

    uri = URI.parse(ENV.fetch("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
    hosts = [{ host: uri.host, port: uri.port || 61613, login: "kubemq", passcode: "" }]
    DESTINATION = "/queue/jobs/email" # Queues pattern → channel jobs.email
    COUNT = 5

    # 1. Produce — SEND `count` jobs to the queue.
    pub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
    (1..COUNT).each { |i| pub.publish(DESTINATION, "job-#{i}", "content-type" => "text/plain") }
    pub.close

    # 2. Consume — SUBSCRIBE ack:client-individual; ACK each delivery by its 1.2 token.
    inbox = Thread::Queue.new
    sub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
    sub.subscribe(DESTINATION, id: "jobs", ack: "client-individual") { |msg| inbox << msg }

    COUNT.times do
      msg = Timeout.timeout(10) { inbox.pop }
      raise "unexpected redelivery" if msg.headers["redelivered"] == "true"

      sub.acknowledge(msg) # ACK by the 1.2 ack token
    end
    sub.close
    puts "drained and acked #{COUNT} jobs"
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use std::time::Duration;

    use async_stomp::client::Connector;
    use async_stomp::{AckMode, FromServer, ToServer};
    use futures::{SinkExt, StreamExt};

    const DESTINATION: &str = "/queue/jobs/email"; // Queues pattern → channel jobs.email
    const COUNT: usize = 5;

    fn host_port() -> (String, u16) {
        let url = std::env::var("KUBEMQ_STOMP_URL").unwrap_or_else(|_| "tcp://localhost:61613".into());
        let hp = url.trim_start_matches("tcp://").trim_start_matches("tls://");
        let mut parts = hp.splitn(2, ':');
        let host = parts.next().unwrap_or("localhost").to_string();
        let port = parts.next().and_then(|p| p.parse().ok()).unwrap_or(61613);
        (host, port)
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let (host, port) = host_port();

        // 1. Produce — SEND `count` jobs to the queue.
        let mut pub_conn = Connector::builder()
            .server(format!("{host}:{port}"))
            .virtualhost(&host)
            .connect()
            .await?;
        for i in 1..=COUNT {
            pub_conn
                .send(ToServer::Send {
                    destination: DESTINATION.into(),
                    transaction: None,
                    headers: Some(vec![("content-type".into(), "text/plain".into())]),
                    body: Some(format!("job-{i}").into_bytes()),
                }.into())
                .await?;
        }
        pub_conn.send(ToServer::Disconnect { receipt: None }.into()).await?;

        // 2. Consume — SUBSCRIBE ack:client-individual; ACK each by its 1.2 token.
        let mut sub = Connector::builder()
            .server(format!("{host}:{port}"))
            .virtualhost(&host)
            .connect()
            .await?;
        sub.send(ToServer::Subscribe {
            destination: DESTINATION.into(),
            id: "jobs".into(),
            ack: Some(AckMode::ClientIndividual),
        }.into())
        .await?;

        let mut got = 0;
        while got < COUNT {
            let frame = tokio::time::timeout(Duration::from_secs(10), sub.next())
                .await?
                .ok_or("stream closed")??;
            if let FromServer::Message { headers, .. } = frame.content {
                let ack_token = headers.iter().find(|(k, _)| k == "ack").map(|(_, v)| v.clone());
                if let Some(token) = ack_token {
                    sub.send(ToServer::Ack { id: token, transaction: None }.into()).await?;
                }
                got += 1;
            }
        }
        println!("drained and acked {COUNT} jobs");
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Delivery guarantees [#delivery-guarantees]

Queues are **at-least-once** — an un-ACKed delivery is **requeued** and redelivered, never lost. Two things trigger a requeue:

* **Ack-timeout (default 30 s).** If a `client-individual` / `client` delivery is not ACKed within the ack-timeout, a 1-second sweeper &#x2A;*NACKs (requeues)** it. The connector does **not** disconnect the client; a late ACK after expiry is silently ignored.
* **Disconnect.** A consumer that disconnects mid-stream with un-ACKed deliveries has all of them NACKed and requeued.

Because dupes are tolerated and never lost, **design consumers to be idempotent**.

Redelivery surfaces **only** via the `redelivered:true` MESSAGE header (broker `ReceiveCount > 1`). There is **no STOMP-level dead-letter queue or redelivery-limit knob** — `maxReceiveCount` / DLQ is broker queue-channel config, not a STOMP feature.

<Callout type="info">
  **SEND closes, SUBSCRIBE gates when the broker is not ready.** If the broker is not ready, a `SEND` returns `ERROR "broker not ready"` and the connection closes (SENDs do not buffer — reconnect and retry). A `SUBSCRIBE` is **gated** instead: accepted and transparently activated when the broker becomes ready (the RECEIPT still fires on acceptance).
</Callout>

## Queue MESSAGE frame fields [#queue-message-frame-fields]

A delivered queue MESSAGE carries: `destination` (always the canonical `/queue/...` form), `message-id` (broker-supplied), `subscription` (**1.1/1.2 only**, omitted on 1.0), an `ack` token (**1.2-only** opaque UUID, distinct from `message-id`), `redelivered:true` when redelivered, custom headers + `content-type` via the `stomp.*` egress mapping, and the body.

## Related [#related]

<Cards>
  <Card title="Events" href="/connectors/stomp/how-to/events" description="Fire-and-forget pub/sub fan-out — every subscriber gets a copy, no ack." />

  <Card title="Commands" href="/connectors/stomp/how-to/commands" description="Requester-only RPC — SEND a command, receive an execution result on a reply inbox." />

  <Card title="Destination grammar" href="/connectors/stomp/reference/destination-grammar" description="The full destination grammar, prefix-to-pattern mapping, and slash-to-dot channel joining." />
</Cards>