# Queues (/connectors/amqp/concepts/queues)



Queues are durable, **competing-consumer** work queues over the AMQP 1.0 connector. Attach a link to a node whose address begins with `queues/<channel>` and the connector binds it to the KubeMQ **Queues** pattern. A message goes to **exactly one** consumer (it is *moved*, not copied), survives until it is settled, and is delivered **at-least-once** by default.

## Overview [#overview]

The `queues/` prefix selects the Queues pattern. A producer attaches a **sender** to `queues/<ch>`; each `TRANSFER` becomes a KubeMQ `SendQueueMessage`. A consumer attaches a **receiver** and grants credit; the server runs a credit-driven `Get` long-poll and frames each returned message as a `TRANSFER`. Do the work, then `accept` — the connector emits an `AckRange` and removes the message from the queue.

Many consumers can attach to the same `queues/<ch>`: the broker hands each message to one of them (competing-consumer move semantics — not fan-out). Add consumers to scale throughput; each message is still processed once.

| Operation | AMQP action                                                                    | KubeMQ mapping                        |
| --------- | ------------------------------------------------------------------------------ | ------------------------------------- |
| Produce   | Attach a **sender** to `queues/<ch>`, `TRANSFER` (unsettled for at-least-once) | `SendQueueMessage`                    |
| Consume   | Attach a **receiver**, grant credit, `Receive`                                 | Credit-driven `Get` long-poll         |
| Accept    | `accepted` / `rejected` DISPOSITION                                            | `AckRange` — message removed          |
| Requeue   | `released` / `modified` DISPOSITION                                            | `NAckRange` — redelivered to the tail |

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

A producer enqueues unsettled messages (each blocks for the server's `accepted` disposition); competing consumers grant credit, receive, do the work, and accept — the message is removed from the queue.

<Mermaid
  chart="`
graph LR
PROD[&#x22;Producer<br/>(sender → queues/tasks)&#x22;]
CONN[&#x22;AMQP 1.0 connector<br/>:5672&#x22;]
BROKER[&#x22;Message Broker&#x22;]
Q{{&#x22;Queue channel<br/>tasks&#x22;}}
W1[&#x22;Worker A&#x22;]
W2[&#x22;Worker B&#x22;]

PROD -- &#x22;TRANSFER (unsettled)&#x22; --> CONN
CONN -- &#x22;SendQueueMessage&#x22; --> BROKER
BROKER --> Q
Q -- &#x22;one message<br/>(move)&#x22; --> W1
Q -. &#x22;next message&#x22; .-> W2
W1 -. &#x22;accepted → AckRange&#x22; .-> CONN

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

*Each queued message is moved to exactly one competing consumer; an `accepted` disposition acks the message and removes it from the queue.*

## Send and receive [#send-and-receive]

Each example produces 10 unsettled messages (every send blocks until the server returns an `accepted` disposition, confirming the broker stored it), then consumes and accepts each one, and finally confirms the queue is empty. Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://localhost:5672`).

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

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

    	amqp "github.com/Azure/go-amqp"
    )

    const channel = "amqp10.examples.basic"
    const total = 10

    func amqpURL() string {
    	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
    		return v
    	}
    	return "amqp://localhost:5672"
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()
    	addr := "queues/" + channel // queues/ prefix → KubeMQ Queues pattern

    	conn, err := amqp.Dial(ctx, amqpURL(), nil)
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	defer func() { _ = conn.Close() }()
    	session, _ := conn.NewSession(ctx, nil)

    	// 1. Produce — each Send is unsettled and blocks for the accepted DISPOSITION.
    	sender, _ := session.NewSender(ctx, addr, nil)
    	for i := 0; i < total; i++ {
    		if err := sender.Send(ctx, amqp.NewMessage([]byte(fmt.Sprintf("msg-%03d", i))), nil); err != nil {
    			log.Fatalf("send: %v", err)
    		}
    	}
    	_ = sender.Close(ctx)

    	// 2. Consume — grant credit, Receive, AcceptMessage (⇒ AckRange, removed).
    	receiver, _ := session.NewReceiver(ctx, addr, &amqp.ReceiverOptions{Credit: 10})
    	seen := make(map[string]struct{}, total)
    	for len(seen) < total {
    		msg, err := receiver.Receive(ctx, nil)
    		if err != nil {
    			log.Fatalf("receive: %v", err)
    		}
    		if err := receiver.AcceptMessage(ctx, msg); err != nil {
    			log.Fatalf("accept: %v", err)
    		}
    		seen[string(msg.GetData())] = struct{}{}
    	}
    	fmt.Printf("consumed and accepted %d messages\n", len(seen))

    	// 3. Assert the queue is empty — a further Receive must time out.
    	emptyCtx, emptyCancel := context.WithTimeout(ctx, 2*time.Second)
    	if _, err := receiver.Receive(emptyCtx, nil); err == nil {
    		log.Fatal("expected an empty queue")
    	}
    	emptyCancel()
    	_ = receiver.Close(ctx)
    }
    ```
  </Tab>

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

    from proton import Message
    from proton.utils import BlockingConnection

    CHANNEL = "amqp10.examples.basic"
    TOTAL = 10


    def amqp_url() -> str:
        return os.environ.get("KUBEMQ_AMQP_URL", "amqp://localhost:5672")


    def main() -> None:
        addr = "queues/" + CHANNEL  # queues/ prefix → KubeMQ Queues pattern
        conn = BlockingConnection(amqp_url())
        try:
            # 1. Produce — each send is unsettled and blocks for the accepted DISPOSITION.
            sender = conn.create_sender(addr)
            for i in range(TOTAL):
                sender.send(Message(body=f"msg-{i:03d}"))
            sender.close()

            # 2. Consume — grant credit, receive, accept (⇒ AckRange, removed).
            receiver = conn.create_receiver(addr, credit=10)
            seen: set[str] = set()
            while len(seen) < TOTAL:
                msg = receiver.receive(timeout=30.0)
                receiver.accept()
                seen.add(str(msg.body))
            print(f"consumed and accepted {len(seen)} messages")

            # 3. Assert the queue is empty — a further receive must time out.
            try:
                receiver.receive(timeout=2.0)
            except Exception:
                pass  # the EXPECTED idle timeout on an empty queue
            else:
                raise SystemExit("expected an empty queue")
            receiver.close()
        finally:
            conn.close()


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

  <Tab value="Java">
    ```java
    import java.util.HashSet;
    import java.util.Set;

    import javax.jms.Connection;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;

    import org.apache.qpid.jms.JmsConnectionFactory;

    public final class Main {
        private static final String CHANNEL = "amqp10.examples.basic";
        private static final int TOTAL = 10;

        public static void main(String[] args) throws Exception {
            String url = System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://localhost:5672");
            String address = "queues/" + CHANNEL; // queues/ prefix → KubeMQ Queues pattern

            JmsConnectionFactory factory = new JmsConnectionFactory(url);
            try (Connection connection = factory.createConnection()) {
                connection.start();
                try (Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE)) {
                    Queue queue = session.createQueue(address);

                    // 1. Produce — a MessageProducer is a server-receiver link; each send
                    //    blocks for the accepted DISPOSITION (unsettled, at-least-once).
                    try (MessageProducer producer = session.createProducer(queue)) {
                        for (int i = 0; i < TOTAL; i++) {
                            producer.send(session.createTextMessage(String.format("msg-%03d", i)));
                        }
                    }

                    // 2. Consume — a CLIENT_ACKNOWLEDGE consumer; acknowledge() settles
                    //    `accepted` ⇒ AckRange (removed from the queue).
                    try (MessageConsumer consumer = session.createConsumer(queue)) {
                        Set<String> seen = new HashSet<>();
                        while (seen.size() < TOTAL) {
                            Message msg = consumer.receive(30_000);
                            if (msg == null) throw new IllegalStateException("timed out");
                            String body = msg.getBody(String.class);
                            msg.acknowledge();
                            seen.add(body);
                        }
                        System.out.printf("consumed and accepted %d messages%n", seen.size());

                        // 3. Assert the queue is empty — a further receive times out.
                        if (consumer.receive(2_000) != null) {
                            throw new IllegalStateException("expected an empty queue");
                        }
                    }
                }
            }
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Text;
    using Amqp;
    using Amqp.Framing;

    const string channel = "amqp10.examples.basic";
    const int total = 10;

    static string AmqpUrl() =>
        Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
            ? v
            : "amqp://localhost:5672";

    var addr = "queues/" + channel; // queues/ prefix → KubeMQ Queues pattern

    var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
    try
    {
        var session = new Session(connection);

        // 1. Produce — each Send is unsettled and blocks for the accepted DISPOSITION.
        var sender = new SenderLink(session, "basic-sender", addr);
        for (var i = 0; i < total; i++)
        {
            var message = new Message { BodySection = new Data { Binary = Encoding.UTF8.GetBytes($"msg-{i:D3}") } };
            sender.Send(message, TimeSpan.FromSeconds(15));
        }

        // 2. Consume — grant credit, Receive, Accept (⇒ AckRange, removed).
        //    Keep the sender open through the consume phase (see the README gotcha).
        var receiver = new ReceiverLink(session, "basic-receiver", addr);
        receiver.SetCredit(10, autoRestore: true);
        var seen = new HashSet<string>();
        while (seen.Count < total)
        {
            var message = receiver.Receive(TimeSpan.FromSeconds(30))
                ?? throw new InvalidOperationException("receive timed out");
            receiver.Accept(message);
            seen.Add(BodyString(message));
        }
        Console.WriteLine($"consumed and accepted {seen.Count} messages");

        // 3. Assert the queue is empty — a further Receive must time out (null).
        if (receiver.Receive(TimeSpan.FromSeconds(2)) is not null)
            throw new InvalidOperationException("expected an empty queue");

        await receiver.CloseAsync();
        await sender.CloseAsync();
        await session.CloseAsync();
    }
    finally
    {
        await connection.CloseAsync();
    }

    static string BodyString(Message message) => message.BodySection switch
    {
        Data d => Encoding.UTF8.GetString(d.Binary),
        AmqpValue { Value: byte[] bytes } => Encoding.UTF8.GetString(bytes),
        AmqpValue { Value: string str } => str,
        AmqpValue v => v.Value?.ToString() ?? string.Empty,
        _ => string.Empty,
    };
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import {
      Connection,
      ReceiverEvents,
      type EventContext,
      type Receiver,
    } from "rhea-promise";

    const channel = "amqp10.examples.basic";
    const total = 10;

    function bodyToString(body: unknown): string {
      return Buffer.isBuffer(body) ? body.toString("utf8") : String(body);
    }

    async function main(): Promise<void> {
      const url = new URL(process.env["KUBEMQ_AMQP_URL"] ?? "amqp://localhost:5672");
      const address = `queues/${channel}`; // queues/ prefix → KubeMQ Queues pattern

      const connection = new Connection({
        host: url.hostname,
        port: url.port ? Number(url.port) : 5672,
        container_id: `kubemq-amqp10-js-basic-${process.pid}`,
        reconnect: false,
      });
      await connection.open();

      try {
        // 1. Produce — an AwaitableSender; each send() resolves on the accepted DISPOSITION.
        const sender = await connection.createAwaitableSender({ target: { address } });
        for (let i = 0; i < total; i++) {
          await sender.send({ body: `msg-${String(i).padStart(3, "0")}` }, { timeoutInSeconds: 15 });
        }
        await sender.close();

        // 2. Consume — grant credit manually, settle manually. Register the handler
        //    BEFORE addCredit so early deliveries are not missed.
        const receiver = await connection.createReceiver({
          source: { address },
          credit_window: 0,
          autoaccept: false,
          autosettle: false,
        });
        const seen = new Set<string>();
        await receiveUntil(receiver, (ctx) => {
          ctx.delivery?.accept(); // accept ⇒ AckRange ⇒ removed from the queue
          seen.add(bodyToString(ctx.message?.body));
          return seen.size >= total;
        }, total + 1, 30_000);
        console.log(`consumed and accepted ${seen.size} messages`);

        await receiver.close();
      } finally {
        await connection.close();
      }
    }

    function receiveUntil(
      receiver: Receiver,
      onMessage: (ctx: EventContext) => boolean,
      credit: number,
      timeoutMs: number,
    ): Promise<void> {
      return new Promise<void>((resolve, reject) => {
        const timer = setTimeout(() => {
          receiver.removeListener(ReceiverEvents.message, handler);
          reject(new Error("timed out waiting for messages"));
        }, timeoutMs);
        const handler = (ctx: EventContext): void => {
          if (onMessage(ctx)) {
            clearTimeout(timer);
            receiver.removeListener(ReceiverEvents.message, handler);
            resolve();
          }
        };
        receiver.on(ReceiverEvents.message, handler);
        receiver.addCredit(credit);
      });
    }

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

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

    use fe2o3_amqp::link::delivery::Delivery;
    use fe2o3_amqp::link::receiver::CreditMode;
    use fe2o3_amqp::{Connection, Receiver, Sender, Session};
    use fe2o3_amqp_types::definitions::SenderSettleMode;
    use fe2o3_amqp_types::messaging::{Body, Message};
    use fe2o3_amqp_types::primitives::Value;

    const CHANNEL: &str = "amqp10.examples.basic";
    const TOTAL: usize = 10;

    fn amqp_url() -> String {
        std::env::var("KUBEMQ_AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_string())
    }

    fn body_bytes(msg: &Message<Body<Value>>) -> Vec<u8> {
        match &msg.body {
            Body::Data(batch) => batch.iter().flat_map(|d| d.0.to_vec()).collect(),
            Body::Value(v) => match &v.0 {
                Value::Binary(b) => b.to_vec(),
                Value::String(s) => s.clone().into_bytes(),
                other => format!("{other:?}").into_bytes(),
            },
            _ => Vec::new(),
        }
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let addr = format!("queues/{CHANNEL}"); // queues/ prefix → KubeMQ Queues pattern

        let mut connection = Connection::open("amqp10-examples-basic", amqp_url().as_str()).await?;
        let mut session = Session::begin(&mut connection).await?;

        // 1. Produce — pin SenderSettleMode::Unsettled (the connector rejects the AMQP
        //    default `mixed`); each send is at-least-once and blocks for the Accepted outcome.
        let mut sender = Sender::builder()
            .name("basic-send-receive-sender")
            .target(addr.as_str())
            .sender_settle_mode(SenderSettleMode::Unsettled)
            .attach(&mut session)
            .await?;
        for i in 0..TOTAL {
            let outcome = sender.send(format!("msg-{i:03}")).await?;
            if !outcome.is_accepted() {
                return Err(format!("send msg-{i:03}: unexpected outcome {outcome:?}").into());
            }
        }
        sender.close().await?;

        // 2. Consume — CreditMode::Auto(10) issues 10 credits and replenishes on settle;
        //    accept ⇒ AckRange ⇒ removed from the queue.
        let mut receiver = Receiver::builder()
            .name("basic-send-receive-receiver")
            .source(addr.as_str())
            .credit_mode(CreditMode::Auto(10))
            .attach(&mut session)
            .await?;
        let mut seen: HashSet<String> = HashSet::with_capacity(TOTAL);
        while seen.len() < TOTAL {
            let delivery: Delivery<Body<Value>> = receiver.recv().await?;
            let body = String::from_utf8_lossy(&body_bytes(delivery.message())).into_owned();
            receiver.accept(&delivery).await?;
            seen.insert(body);
        }
        println!("consumed and accepted {} messages", seen.len());

        // 3. Assert the queue is empty — a further recv must time out.
        if let Ok(Ok(_)) = tokio::time::timeout(Duration::from_secs(2), receiver.recv::<Body<Value>>()).await {
            return Err("expected an empty queue".into());
        }

        receiver.close().await?;
        session.end().await?;
        connection.close().await?;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Delivery guarantees [#delivery-guarantees]

Settlement mode is negotiated at `ATTACH` and decides the produce/consume guarantee:

| Mode                           | How                                                        | Guarantee                                                            | Use when                     |
| ------------------------------ | ---------------------------------------------------------- | -------------------------------------------------------------------- | ---------------------------- |
| **At-least-once (default)**    | unsettled deliveries; you `accept` after the work succeeds | survives a crash; may be **redelivered**                             | the work must not be lost    |
| **Pre-settled (at-most-once)** | `snd-settle-mode=settled` on the sender / consume          | fast, no DISPOSITION round-trip; a failed publish/consume is dropped | loss is acceptable for speed |

At-least-once means a consumer crash before `accept` requeues the message, so workers **must be idempotent** — a message can arrive more than once. The receiver settle mode is always replied as `first`; requesting `second` returns `DETACH(amqp:not-implemented)`.

### Delivery-state outcome mapping [#delivery-state-outcome-mapping]

When you consume, you settle each delivery by sending a `DISPOSITION` with a delivery state. The connector maps that state onto the KubeMQ queue Ack/NAck machinery:

| Your DISPOSITION                | Client call (typical)  | KubeMQ action | Effect                                                                        |
| ------------------------------- | ---------------------- | ------------- | ----------------------------------------------------------------------------- |
| `accepted`                      | `AcceptMessage`        | **AckRange**  | message removed                                                               |
| `rejected`                      | `RejectMessage`        | **AckRange**  | discarded; poison handled by the broker's `MaxReceiveQueue` policy            |
| `released`                      | `ReleaseMessage`       | **NAckRange** | redelivered to the tail; `delivery-count` grows; **increments receive-count** |
| `modified{...}`                 | `ModifyMessage`        | **NAckRange** | requeued to the tail                                                          |
| nil state (settled, no outcome) | settle without a state | **AckRange**  | treated as success                                                            |
| unknown terminal state          | —                      | **NAckRange** | conservatively requeued — never silently dropped                              |

A redelivered message carries `header.delivery-count = ReceiveCount − 1` and `first-acquirer = (ReceiveCount == 1)` — use these to detect and de-duplicate redeliveries. On detach, connection close, or shutdown, **every unsettled delivery is NAcked exactly once** (returned to the queue tail), so a disconnecting worker loses nothing — a fresh consumer recovers the in-flight work.

<Callout type="warn">
  **`released` / `modified` increment the receive-count.** Every release/modify for redelivery bumps `ReceiveCount` toward the broker's `MaxReceiveQueue` cap. A message you keep NAcking eventually hits that cap and is removed **even though you never `rejected` it** — there is no requeue-without-increment. To genuinely discard, `reject`; to retry, understand the count climbs.
</Callout>

### Body sections [#body-sections]

A message body must be `Data` (binary — the default; multiple `Data` sections concatenate) or `AmqpValue` (a typed value). An empty body is valid. An `AmqpSequence` body is **rejected** (`rejected` DISPOSITION then `DETACH(amqp:not-implemented)`).

## What queues do not have [#what-queues-do-not-have]

The Queues pattern is deliberately minimal — none of these exist:

* **No peek / browse / visibility-timeout.** Receive is destructive credit-based consume only; there is no "look without taking".
* **No connector dead-letter exchange.** A `rejected` message is discarded; poison handling is the broker-side `MaxReceiveQueue` policy, not a per-link DLX.
* **No `copy` distribution-mode.** Requesting `copy` on a `queues/` link returns `DETACH(amqp:invalid-field)` — queues are move-only.
* **No selectors.** A selector filter on a `queues/` link returns `amqp:not-implemented` (selectors are pub/sub only).
* **No transactions.** Reliability comes from settlement, not `SESSION_TRANSACTED`.

## Related [#related]

<Cards>
  <Card title="Events" href="/connectors/amqp/concepts/events" description="Fire-and-forget pub/sub fan-out — every subscriber gets a copy." />

  <Card title="Commands" href="/connectors/amqp/concepts/commands" description="Native request/reply with a positive failure signal (executed=false)." />

  <Card title="Address mapping" href="/connectors/amqp/reference/address-mapping" description="The full address grammar, longest-prefix matching, and the queues/ prefix." />
</Cards>