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



Get a message flowing through the KubeMQ AMQP 1.0 connector in minutes. You point a
standard AMQP 1.0 client at the broker, attach a sender to `queues/<channel>`, produce a
few messages, then attach a receiver and consume them back — all over the native AMQP 1.0
wire, with no KubeMQ SDK. This walkthrough takes you from a running server to a verified
at-least-once round-trip.

## Prerequisites [#prerequisites]

* A running **kubemq-server** with the AMQP 1.0 connector **enabled** and reachable on **port 5672**
  (plain TCP). The connector is &#x2A;*opt-in (disabled by default)** — see the enable step below.
* One of the AMQP 1.0 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 AMQP 1.0 connector is **disabled by default** — a stock kubemq-server does **not** bind
the AMQP 1.0 listener until you turn it on. Enable it with its enable variable:

<RunKubeMQ ports="[5672, 5671, 50000]" env="{ CONNECTORS_AMQP10_ENABLE: 'true' }" />

<Callout type="warn">
  The enable variable is &#x2A;*`CONNECTORS_AMQP10_ENABLE`** — the literal `10` stays attached to
  `AMQP` with no underscore. `CONNECTORS_AMQP_1_0_ENABLE` and `CONNECTORS_AMQP10ENABLE` do
  **not** bind. For Kubernetes, set `spec.amqp10.enabled: true` in the `KubemqCluster` CR.
</Callout>

Bring up a throwaway local broker with AMQP 1.0 enabled:

Every example reads a single environment variable for the broker endpoint. A URL with no
userinfo negotiates **SASL ANONYMOUS**, so a stock dev broker is clone-and-run with no
credentials:

```bash
# default: amqp://localhost:5672
export KUBEMQ_AMQP_URL="amqp://localhost:5672"
```

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

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

When `Enable` is `false`, all other AMQP 1.0 validation is skipped and no listener binds. See
[Configuration](/connectors/amqp/concepts/configuration) for the full settings list.

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

A producer attaches a sender link to a node address; the connector resolves the address
prefix to a KubeMQ pattern and channel and enqueues each message. A consumer attaches a
receiver link to the same address, grants credit, and the connector delivers and removes
each accepted message.

<Mermaid
  chart="`
graph LR
PROD[&#x22;Producer<br/>(sender link)&#x22;]
CONN[&#x22;AMQP 1.0 connector<br/>:5672&#x22;]
CH{{&#x22;Queue channel<br/>queues/&lt;ch&gt;&#x22;}}
CONS[&#x22;Consumer<br/>(receiver link)&#x22;]

PROD -- &#x22;TRANSFER (unsettled)&#x22; --> CONN
CONN -- &#x22;enqueue&#x22; --> CH
CH -. &#x22;deliver on credit&#x22; .-> CONN
CONN -. &#x22;TRANSFER → accept&#x22; .-> CONS

class PROD,CONS client
class CONN connector
class CH queues
`"
/>

*A sender produces unsettled messages to a Queue channel; a receiver grants credit, and the connector delivers and removes each accepted message.*

## Steps [#steps]

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

    Open an AMQP 1.0 connection to the endpoint in `KUBEMQ_AMQP_URL`. A URL with no userinfo
    negotiates SASL ANONYMOUS; the client sends a non-empty `container-id` automatically (the
    connector requires one). One session carries both the producer and consumer links in the
    steps below.

    The language tabs across all three steps run the **complete** round-trip from a single
    program: connect, produce 10 messages to `queues/amqp10.examples.basic`, consume and
    accept each, and confirm the queue drains to empty.

    <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

        	// CONNECT: OPEN (SASL ANONYMOUS) + BEGIN one session.
        	conn, err := amqp.Dial(ctx, amqpURL(), nil)
        	if err != nil {
        		log.Fatalf("dial: %v", err)
        	}
        	defer func() { _ = conn.Close() }()
        	session, err := conn.NewSession(ctx, nil)
        	if err != nil {
        		log.Fatalf("new session: %v", err)
        	}

        	// SEND: attach a sender; each unsettled Send blocks for the accepted disposition.
        	sender, err := session.NewSender(ctx, addr, nil)
        	if err != nil {
        		log.Fatalf("new sender: %v", err)
        	}
        	for i := 0; i < total; i++ {
        		body := fmt.Sprintf("msg-%03d", i)
        		if err := sender.Send(ctx, amqp.NewMessage([]byte(body)), nil); err != nil {
        			log.Fatalf("send %s: %v", body, err)
        		}
        	}
        	_ = sender.Close(ctx)
        	fmt.Printf("[send] Produced %d messages to %s\n", total, addr)

        	// RECEIVE: attach a receiver with credit; accept each => removed from the queue.
        	receiver, err := session.NewReceiver(ctx, addr, &amqp.ReceiverOptions{Credit: 10})
        	if err != nil {
        		log.Fatalf("new receiver: %v", err)
        	}
        	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("[recv] Consumed and accepted %d messages (no loss)\n", len(seen))
        	_ = 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

            # CONNECT: OPEN (SASL ANONYMOUS) — proton sends a non-empty container-id.
            conn = BlockingConnection(amqp_url())
            try:
                # SEND: attach a sender; each send 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()
                print(f"[send] Produced {TOTAL} messages to {addr}")

                # RECEIVE: attach a receiver with credit; accept each => removed from the queue.
                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"[recv] Consumed and accepted {len(seen)} messages (no loss)")
                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 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;

                // CONNECT: OPEN (SASL ANONYMOUS). The JMS destination name IS the node address.
                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);

                        // SEND: a producer is a server-receiver link; send blocks for accepted.
                        try (MessageProducer producer = session.createProducer(queue)) {
                            for (int i = 0; i < TOTAL; i++) {
                                producer.send(session.createTextMessage(String.format("msg-%03d", i)));
                            }
                        }
                        System.out.printf("[send] Produced %d messages to %s%n", TOTAL, address);

                        // RECEIVE: a CLIENT_ACKNOWLEDGE consumer; acknowledge() settles accepted.
                        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 before draining the queue");
                                }
                                String body = msg.getBody(String.class);
                                msg.acknowledge();
                                seen.add(body);
                            }
                            System.out.printf("[recv] Consumed and accepted %d messages (no loss)%n", seen.size());
                        }
                    }
                }
            }
        }
        ```
      </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;

        // CONNECT: OPEN (SASL ANONYMOUS) + one session.
        var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
        try
        {
            var session = new Session(connection);

            // SEND: a SenderLink is a server-receiver link; Send blocks for accepted.
            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));
            }
            Console.WriteLine($"[send] Produced {total} messages to {addr}");

            // RECEIVE: grant credit; Accept each => removed from the queue. Keep the
            // sender open through the consume phase (detaching it early can stall delivery).
            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($"[recv] Consumed and accepted {seen.Count} messages (no loss)");

            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 ConnectionOptions,
          type EventContext,
          type Receiver,
        } from "rhea-promise";

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

        function connectionOptions(): ConnectionOptions {
          const url = new URL(process.env["KUBEMQ_AMQP_URL"] ?? "amqp://localhost:5672");
          return {
            host: url.hostname,
            port: url.port ? Number(url.port) : 5672,
            container_id: `kubemq-amqp10-js-${process.pid}`,
            reconnect: false,
          };
        }

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

        async function main(): Promise<void> {
          const address = `queues/${channel}`;

          // CONNECT: OPEN (SASL ANONYMOUS).
          const connection = new Connection(connectionOptions());
          await connection.open();
          try {
            // SEND: 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();
            console.log(`[send] Produced ${total} messages to ${address}`);

            // RECEIVE: manual credit + manual settle; accept() removes from the queue.
            const receiver = await connection.createReceiver({
              source: { address },
              credit_window: 0,
              autoaccept: false,
              autosettle: false,
            });
            const seen = new Set<string>();
            await new Promise<void>((resolve, reject) => {
              receiver.on(ReceiverEvents.message, (ctx: EventContext) => {
                ctx.delivery?.accept();
                seen.add(bodyToString(ctx.message?.body));
                if (seen.size >= total) resolve();
              });
              receiver.on(ReceiverEvents.receiverError, (ctx: EventContext) =>
                reject(ctx.receiver?.error ?? new Error("receiver error")));
              (receiver as Receiver).addCredit(total);
            });
            console.log(`[recv] Consumed and accepted ${seen.size} messages (no loss)`);
            await receiver.close();
          } finally {
            await connection.close();
          }
        }

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

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

        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_string(msg: &Message<Body<Value>>) -> String {
            match &msg.body {
                Body::Data(batch) => batch.iter().flat_map(|d| d.0.iter().copied()).collect::<Vec<u8>>(),
                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(),
            }
            .into_iter()
            .map(|b| b as char)
            .collect()
        }

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

            // CONNECT: OPEN (SASL ANONYMOUS) + BEGIN one session.
            let mut connection = Connection::open("amqp10-examples-basic", amqp_url().as_str()).await?;
            let mut session = Session::begin(&mut connection).await?;

            // SEND: a sender pinned to Unsettled (at-least-once); the default `mixed` is rejected.
            let mut sender = Sender::builder()
                .name("basic-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 {i}: unexpected outcome {outcome:?}").into());
                }
            }
            sender.close().await?;
            println!("[send] Produced {TOTAL} messages to {addr}");

            // RECEIVE: client-granted auto credit; accept each => removed from the queue.
            let mut receiver = Receiver::builder()
                .name("basic-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?;
                receiver.accept(&delivery).await?;
                seen.insert(body_string(delivery.message()));
            }
            println!("[recv] Consumed and accepted {} messages (no loss)", seen.len());

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

  <Step>
    ### Send messages [#send-messages]

    The send phase in the program above attaches a **sender** link with the target address
    `queues/amqp10.examples.basic` and produces 10 messages. Because the sends are *unsettled*,
    each call blocks until the connector returns an `accepted` disposition — confirmation that
    the broker stored the message (at-least-once). Always use the **explicit pattern prefix**
    (`queues/…`); never rely on the connector's bare-address fallback.
  </Step>

  <Step>
    ### Receive and verify [#receive-and-verify]

    The receive phase attaches a **receiver** link to the same address and grants link credit;
    the connector delivers each message, and `accept` settles it — emitting an `AckRange` that
    removes it from the queue. After consuming all 10, a further receive times out, proving the
    queue drained with no loss:

    ```text
    [send] Produced 10 messages to queues/amqp10.examples.basic
    [recv] Consumed and accepted 10 messages (no loss)
    [recv] Queue drained to empty (no further messages)
    ```

    Behind that, the full AMQP 1.0 handshake ran end-to-end: **OPEN** → **BEGIN** →
    **ATTACH** (sender, then receiver) → **TRANSFER / FLOW / DISPOSITION** → **DETACH /
    CLOSE**.
  </Step>
</Steps>

<Callout type="info">
  Two flow-control footguns to internalize early: **Events** drop silently at zero credit
  (keep a consumer's credit topped up), and **Events-Store** stalls and loses its window if
  a durable consumer's credit is not replenished. Both are covered in
  [Flow control](/connectors/amqp/how-to/flow-control).
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Configuration" href="/connectors/amqp/concepts/configuration" description="The 14 connector settings, the CONNECTORS_AMQP10_ENABLE disable var, defaults, and validation." />

  <Card title="Architecture" href="/connectors/amqp/concepts/architecture" description="The amqpmux front door, the connection/session/link model, and the metadata envelope." />

  <Card title="Queues" href="/connectors/amqp/concepts/queues" description="At-least-once enqueue and credit-driven destructive consume over AMQP 1.0." />

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