# AMQP 1.0 (/connectors/amqp)



Point your AMQP 1.0 application at KubeMQ by changing only the connection string and
node address. The **AMQP 1.0 connector** is a built-in, wire-protocol bridge inside
kubemq-server that speaks the OASIS AMQP 1.0 dialect natively — any standard AMQP 1.0
client (Qpid, go-amqp, AMQPNetLite, rhea) talks to KubeMQ's Queues, Events,
Events-Store, Commands, and Queries with no KubeMQ SDK, no library swap, and no code
rewrite.

## What is the AMQP 1.0 connector [#what-is-the-amqp-10-connector]

The AMQP 1.0 connector exposes [AMQP 1.0](https://www.amqp.org/) (OASIS / ISO-IEC 19464)
natively over a dedicated port for **all five KubeMQ messaging patterns**. The leading
segment of the address a client attaches to selects the pattern: attaching to
`queues/orders` binds to a KubeMQ Queue, `events/telemetry` to Events, and so on. The
connector is a *gateway*, not a client library — your application only needs a stock
AMQP 1.0 client.

<Callout type="info">
  AMQP 1.0 is a peer-to-peer [link](/connectors/amqp/concepts/architecture#connection--session--link)
  protocol, distinct from the AMQP 0-9-1 dialect the [RabbitMQ connector](/connectors/rabbitmq)
  speaks: a link attaches to a *node* (an address), flow is governed by *credit*, and
  delivery is resolved by a *delivery state* (accepted / released / modified / rejected) —
  there are no exchanges, bindings, routing keys, or publisher-confirms. See
  [Architecture](/connectors/amqp/concepts/architecture) for the full model, or
  [Migrating from ActiveMQ](/connectors/how-to/migration/from-activemq) if you're
  coming from 0-9-1.
</Callout>

Key capabilities:

* **All five patterns over one wire** — Queues, Events, Events-Store, Commands, and
  Queries, selected by the address prefix.
* **Address-driven pattern routing** — `queues/`, `events/`, `events-store/`,
  `commands/`, `queries/` prefixes map a link to a KubeMQ pattern by longest-prefix
  match.
* **Native settlement and credit** — at-least-once (unsettled) or at-most-once
  (pre-settled) delivery, credit-driven flow control, and the standard AMQP dispositions.
* **Cross-protocol interop** — a message sent over AMQP 1.0 to `queues/orders` is
  consumable by a gRPC or REST KubeMQ client on the same channel, and vice-versa.

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

An AMQP 1.0 client connects to the connector and attaches a link to a node address. The
connector resolves the address to a KubeMQ `(pattern, channel)` pair, hands the message to
the message broker, and consumers on the same channel — over AMQP 1.0 or any other KubeMQ
transport — receive it.

<Mermaid
  chart="`
graph LR
APP[&#x22;AMQP 1.0 client<br/>(any language)&#x22;]
MUX[&#x22;amqpmux<br/>:5672 / :5671&#x22;]
CONN[&#x22;AMQP 1.0 connector&#x22;]
BROKER[&#x22;Message Broker&#x22;]
SUB[&#x22;Consumer<br/>(AMQP 1.0 / gRPC / REST)&#x22;]

APP -- &#x22;ATTACH queues/orders&#x22; --> MUX
MUX -- &#x22;8-byte header → 1.0&#x22; --> CONN
CONN -- &#x22;(pattern, channel)&#x22; --> BROKER
BROKER -. &#x22;deliver&#x22; .-> SUB

class APP,SUB client
class MUX,CONN connector
class BROKER broker
`"
/>

*The shared `amqpmux` front door classifies the connection by its 8-byte protocol header and dispatches it to the AMQP 1.0 engine, which maps the node address onto a KubeMQ pattern and channel.*

## Ports & protocol surface [#ports--protocol-surface]

| Port   | Transport                          | Protocol                         | Notes                                                                                                   |
| ------ | ---------------------------------- | -------------------------------- | ------------------------------------------------------------------------------------------------------- |
| `5672` | Plain TCP (SASL ANONYMOUS / PLAIN) | AMQP 1.0 (OASIS / ISO-IEC 19464) | Default plain listener. **Shared with the RabbitMQ (AMQP 0-9-1) connector** via the internal `amqpmux`. |
| `5671` | TLS over TCP                       | AMQP 1.0                         | Binds only when the server-global `Security` block is configured. Shared TLS listener with AMQP 0-9-1.  |

A single `amqpmux` listener accepts every connection on `5672`/`5671`, reads the 8-byte
AMQP protocol header, and routes it to the matching dialect engine — so AMQP 1.0 and
AMQP 0-9-1 coexist on the same ports. There is **no vhost**: the OPEN `hostname` field is
accepted and ignored. See [Architecture](/connectors/amqp/concepts/architecture) for the
dispatch detail.

## Send a message [#send-a-message]

The example below produces one message to `queues/<channel>` over a stock AMQP 1.0 client.
Each send is unsettled (at-least-once): it blocks until the connector returns an `accepted`
disposition, confirming the broker stored the message. 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"
    )

    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(), 30*time.Second)
    	defer cancel()

    	addr := "queues/amqp10.examples.basic"

    	// OPEN: SASL ANONYMOUS by default (no userinfo in the URL).
    	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)
    	}

    	// ATTACH a sender (server-receiver link) and send one unsettled message.
    	sender, err := session.NewSender(ctx, addr, nil)
    	if err != nil {
    		log.Fatalf("new sender: %v", err)
    	}
    	if err := sender.Send(ctx, amqp.NewMessage([]byte("hello from AMQP 1.0")), nil); err != nil {
    		log.Fatalf("send: %v", err)
    	}
    	_ = sender.Close(ctx)
    	fmt.Printf("sent 1 message to %s (accepted)\n", addr)
    }
    ```
  </Tab>

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

    from proton import Message
    from proton.utils import BlockingConnection


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


    def main() -> None:
        addr = "queues/amqp10.examples.basic"

        # OPEN: SASL ANONYMOUS by default (no userinfo in the URL).
        conn = BlockingConnection(amqp_url())
        try:
            # ATTACH a sender; each send blocks for the accepted disposition.
            sender = conn.create_sender(addr)
            sender.send(Message(body="hello from AMQP 1.0"))
            sender.close()
            print(f"sent 1 message to {addr} (accepted)")
        finally:
            conn.close()


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

  <Tab value="Java">
    ```java
    import javax.jms.Connection;
    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 {
        public static void main(String[] args) throws Exception {
            String url = System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://localhost:5672");
            String address = "queues/amqp10.examples.basic";

            // OPEN: SASL ANONYMOUS by default. The JMS destination name IS the
            // connector 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);
                    try (MessageProducer producer = session.createProducer(queue)) {
                        TextMessage msg = session.createTextMessage("hello from AMQP 1.0");
                        producer.send(msg); // blocks until the accepted DISPOSITION
                    }
                    System.out.printf("sent 1 message to %s (accepted)%n", address);
                }
            }
        }
    }
    ```
  </Tab>

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

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

    var addr = "queues/amqp10.examples.basic";

    // OPEN: SASL ANONYMOUS by default (no userinfo in the URL).
    var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
    try
    {
        var session = new Session(connection);

        // ATTACH a sender (server-receiver link) and send one unsettled message.
        var sender = new SenderLink(session, "basic-sender", addr);
        var message = new Message
        {
            BodySection = new Data { Binary = Encoding.UTF8.GetBytes("hello from AMQP 1.0") },
        };
        sender.Send(message, TimeSpan.FromSeconds(15)); // blocks for the accepted DISPOSITION
        await sender.CloseAsync();
        await session.CloseAsync();
        Console.WriteLine($"sent 1 message to {addr} (accepted)");
    }
    finally
    {
        await connection.CloseAsync();
    }
    ```
  </Tab>

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

    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,
        // The connector requires a non-empty container-id; rhea sends one by default.
        container_id: `kubemq-amqp10-js-${process.pid}`,
        reconnect: false,
      };
    }

    async function main(): Promise<void> {
      const address = "queues/amqp10.examples.basic";

      // OPEN: SASL ANONYMOUS by default (no username/password).
      const connection = new Connection(connectionOptions());
      await connection.open();
      try {
        // Attach an AwaitableSender; send() resolves on the accepted disposition.
        const sender = await connection.createAwaitableSender({ target: { address } });
        await sender.send({ body: "hello from AMQP 1.0" }, { timeoutInSeconds: 15 });
        await sender.close();
        console.log(`sent 1 message to ${address} (accepted)`);
      } finally {
        await connection.close();
      }
    }

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

  <Tab value="Rust">
    ```rust
    use fe2o3_amqp::{Connection, Sender, Session};
    use fe2o3_amqp_types::definitions::SenderSettleMode;

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

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let addr = "queues/amqp10.examples.basic";

        // OPEN: SASL ANONYMOUS by default; a non-empty container-id is required.
        let mut connection = Connection::open("kubemq-amqp10-rust", amqp_url().as_str()).await?;
        let mut session = Session::begin(&mut connection).await?;

        // ATTACH a sender pinned to Unsettled (at-least-once); the connector
        // rejects the AMQP default `mixed`.
        let mut sender = Sender::builder()
            .name("basic-sender")
            .target(addr)
            .sender_settle_mode(SenderSettleMode::Unsettled)
            .attach(&mut session)
            .await?;

        let outcome = sender.send("hello from AMQP 1.0").await?;
        if !outcome.is_accepted() {
            return Err(format!("unexpected outcome {outcome:?}").into());
        }
        sender.close().await?;
        println!("sent 1 message to {addr} (accepted)");

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

## Supported languages [#supported-languages]

The connector speaks standard AMQP 1.0, so any conformant client works. The examples
pin one native AMQP 1.0 client per language — there is no KubeMQ SDK, no proto bindings,
and no published package.

| Language                | Client library                                                                                  | Notes                             |
| ----------------------- | ----------------------------------------------------------------------------------------------- | --------------------------------- |
| Go                      | [`github.com/Azure/go-amqp`](https://github.com/Azure/go-amqp)                                  | The connector's reference client. |
| Python                  | [`python-qpid-proton`](https://qpid.apache.org/proton/)                                         | Sync `BlockingConnection`.        |
| Java                    | [`org.apache.qpid:qpid-jms-client`](https://qpid.apache.org/components/jms/)                    | `javax.jms` (not Jakarta).        |
| C# / .NET               | [`AMQPNetLite.Core`](https://github.com/Azure/amqpnetlite)                                      | Task-based async.                 |
| JavaScript / TypeScript | [`rhea`](https://github.com/amqp/rhea) + [`rhea-promise`](https://github.com/amqp/rhea-promise) | Event-driven, promise-wrapped.    |
| Rust                    | [`fe2o3-amqp`](https://github.com/minghuaw/fe2o3-amqp)                                          | async/await on Tokio.             |

<Callout type="info">
  Apache Qpid JMS (Java) cannot drive the anonymous-terminus link — it has no API to force
  a raw null-target link, and the connector advertises no `ANONYMOUS-RELAY` capability. Use
  per-pattern senders instead. See [Address mapping](/connectors/amqp/reference/address-mapping).
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Getting started" href="/connectors/amqp/tutorials/getting-started" description="Connect, send, and receive a message end-to-end through the AMQP 1.0 connector in minutes." />

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

  <Card title="Events" href="/connectors/amqp/concepts/events" description="Pub/sub fan-out over AMQP 1.0 — pre-settled delivery, credit, and consumer groups." />

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