Migrating from AMQP 1.0

If you have an application that speaks native AMQP 1.0 (ISO/IEC 19464) — using a client such as Azure/go-amqp, AMQP.NET Lite, or Apache Qpid Proton — you can point it at KubeMQ's built-in AMQP 1.0 connector by changing only the endpoint and, where required, the SASL credentials. Address prefixes select the KubeMQ messaging pattern, so simple publish/consume needs no app-code rewrite. This is a drop-in migration at the endpoint / client level.

If you are migrating a JMS application (Java) instead, see the Migrating from JMS guide; for ActiveMQ applications routed by client type, see Migrating from ActiveMQ.

Overview

The AMQP 1.0 connector exposes KubeMQ's Queues, Events, Events Store, Commands, and Queries patterns over the native AMQP 1.0 wire protocol. It listens on the same ports as the AMQP 0-9-1 connector — 5672 (plain / SASL) and 5671 (TLS / mTLS) — because both protocols share a single listener that routes each connection by its protocol header. No separate firewall rule is needed beyond what the AMQP port already allows.

• Canonical client (this guide): github.com/Azure/go-amqp v1.7.0. For .NET shops, AMQP.NET Lite is a direct alternative; the connection-string and address conventions are the same, but the snippets below target go-amqp.
• Opt-in default: The connector is disabled by default. Set CONNECTORS_AMQP10_ENABLE=true (or Enable = true under [Connectors.Amqp10] in TOML) before connecting.

Enable the connector before you migrate any traffic:

docker run -d \  --name kubemq \  -p 5672:5672 \  -p 5671:5671 \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  -e CONNECTORS_AMQP10_ENABLE=true \  europe-docker.pkg.dev/kubemq/images/kubemq:next

For the full wire-protocol contract, see the AMQP connector capabilities reference.

Compatibility Matrix

The cells below describe the AMQP 1.0 column of the cross-protocol migration matrix.

¹ There is no client-settable DLQ over this protocol. The AMQP 1.0 connector never marks published messages for dead-lettering, so a poison message that exceeds MaxReceiveCount is silently dropped by the broker rather than delivered to any consumable dead-letter address. For a genuine client-facing DLQ, use the RabbitMQ (DLX) or AWS (redrive) connector.

Connection / Endpoint Migration

Change only the host and, if required, the SASL credentials. The AMQP 1.0 port is shared with AMQP 0-9-1 on the same listener, so no firewall change is needed beyond what the AMQP port already allows.

// go-amqp v1.7.0 — amqp.Dial
conn, err := amqp.Dial(ctx, "amqp://kubemq-host:5672",
    &amqp.ConnOptions{
        SASLType: amqp.SASLTypePlain("svc-orders", "<kubemq-jwt>"),
    })

When Authentication.Enable = false on the server, the connector also accepts SASL ANONYMOUS and bare AMQP headers (no SASL) — convenient for local development.

Concept & Destination Mapping

The address prefix of the AMQP link selects the KubeMQ messaging pattern. The leading / is optional (queues/orders ≡ /queues/orders).

Selectors (Events / Events Store only): attach a filter under the apache.org:selector-filter:string descriptor on the receiver link source. The SQL92 subset supported includes comparisons, AND / OR / NOT, BETWEEN, IN, LIKE, IS NULL, and parentheses, evaluated against application-properties (and standard JMS headers).

Bare addresses: when no prefix is present the connector resolves by JMS terminus capability hint (queue → Queues, topic → Events) or falls back to DefaultPattern (default: "queues").

Interop with AMQP 0-9-1: channels produced over the AMQP 0-9-1 connector use the prefix amqp.<vhost>.<queue>; an AMQP 1.0 client reaches the same data at /queues/amqp.<vhost>.<queue>.

See the address mapping reference for the full grammar and the longest-prefix rule.

From other AMQP 1.0 brokers (Solace / Azure Service Bus)

The connector speaks standard AMQP 1.0, so non-ActiveMQ AMQP 1.0 clients migrate the same way — change the endpoint, then map destinations to <pattern>/<channel>.

• Solace PubSub+ — a Solace AMQP 1.0 sender/receiver targets a queue or topic by name. Remap the Solace destination to queues/<name> (persistent) or events/<name> (direct). Solace exclusive/non-exclusive durable topic endpoints map to events-store/<name> durable subscriptions. Solace selectors map onto the pub/sub selector (events/-only).
• Azure Service Bus — Service Bus AMQP 1.0 entities (queues/<q>, topics/<t>/subscriptions/<s>) remap to queues/<channel> and events-store/<channel> (durable). Azure SB sessions, scheduled/deferred delivery, dead-lettering, and transactions have no KubeMQ equivalent — drop those features (see Capabilities). Azure SB's amqps:// + SAS-token auth maps to KubeMQ SASL PLAIN with a JWT.

For any AMQP 1.0 broker, the discipline is identical: explicit <pattern>/<channel> addresses, continuous credit for at-most-once patterns, symbolic amqp:* error conditions (never numeric codes), and the deviations below.

Canonical Client Example

Client: github.com/Azure/go-amqp v1.7.0 Symbols used: amqp.Dial, conn.NewSession, session.NewSender, session.NewReceiver, sender.Send, amqp.NewMessage, receiver.Receive, receiver.AcceptMessage, plus msg.Properties.ReplyTo / msg.Properties.CorrelationID for RPC.

Publish and consume (Queues)

package main

import (
    "context"
    "fmt"
    "log"

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

func main() {
    ctx := context.Background()

    // amqp.Dial establishes the TCP connection and SASL handshake.
    conn, err := amqp.Dial(ctx, "amqp://kubemq-host:5672",
        &amqp.ConnOptions{
            SASLType: amqp.SASLTypePlain("svc-orders", "<kubemq-jwt>"),
        })
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    // conn.NewSession opens an AMQP session.
    sess, err := conn.NewSession(ctx, nil)
    if err != nil {
        log.Fatal(err)
    }

    // --- Publish ---
    // session.NewSender attaches a sender link to /queues/orders.
    snd, err := sess.NewSender(ctx, "/queues/orders", nil)
    if err != nil {
        log.Fatal(err)
    }
    // sender.Send transfers one message; amqp.NewMessage wraps the body.
    if err := snd.Send(ctx, amqp.NewMessage([]byte(`{"id":"1","item":"widget"}`)), nil); err != nil {
        log.Fatal(err)
    }
    snd.Close(ctx)

    // --- Consume ---
    // session.NewReceiver attaches a competing-consumer receiver on the same queue.
    rcv, err := sess.NewReceiver(ctx, "/queues/orders", &amqp.ReceiverOptions{
        Credit: 10, // grant initial link credit
    })
    if err != nil {
        log.Fatal(err)
    }
    // receiver.Receive blocks until a message arrives.
    msg, err := rcv.Receive(ctx, nil)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("received: %s\n", msg.GetData())
    // receiver.AcceptMessage settles the delivery (DISPOSITION accepted → broker AckRange).
    if err := rcv.AcceptMessage(ctx, msg); err != nil {
        log.Fatal(err)
    }
    rcv.Close(ctx)
}

Events Store (durable subscription)

// session.NewReceiver on /events-store/<ch> → durable Events Store subscription.
// The broker resumes from the last acknowledged position on reconnect.
rcv, err := sess.NewReceiver(ctx, "/events-store/audit", &amqp.ReceiverOptions{
    Credit: 64,
    Durability: amqp.DurabilityUnsettledState, // terminus expiry-policy 'never'
})

Selectors (Events / Events Store)

// Attach a SQL92 selector on an events receiver link source filter.
// Selector is evaluated in the connector before delivery.
rcv, err := sess.NewReceiver(ctx, "/events/orders",
    &amqp.ReceiverOptions{
        Credit: 32,
        Filters: []amqp.LinkFilter{
            amqp.NewSelectorFilter("priority > 5 AND region = 'EU'"),
        },
    })

RPC — hand-rolled reply receiver

The AMQP 1.0 connector has no library-level request/reply helper. You must create a dynamic reply receiver yourself, set msg.Properties.ReplyTo to its address, and match responses by CorrelationID. The example below uses the exact go-amqp v1.7.0 symbols.

// 1. Open a dynamic receiver to serve as the reply address.
//    session.NewReceiver with DynamicAddress=true → connector allocates
//    a temporary node and returns its address in the ATTACH reply.
replyRcv, err := sess.NewReceiver(ctx, "", &amqp.ReceiverOptions{
    Credit:         1,
    DynamicAddress: true,
})
if err != nil {
    log.Fatal(err)
}
replyAddr := replyRcv.Address() // the connector-assigned dynamic node address

// 2. Attach a sender to the command channel.
snd, err := sess.NewSender(ctx, "/commands/status", nil)
if err != nil {
    log.Fatal(err)
}

// 3. Build the request message.
//    msg.Properties.ReplyTo tells the connector where to send the response.
//    msg.Properties.CorrelationID allows matching the reply to the request.
req := amqp.NewMessage([]byte(`{"service":"inventory"}`))
req.Properties = &amqp.MessageProperties{
    ReplyTo:       &replyAddr,
    CorrelationID: "req-001",
}

// sender.Send dispatches the request to the Commands channel.
if err := snd.Send(ctx, req, nil); err != nil {
    log.Fatal(err)
}

// 4. receiver.Receive blocks for the reply; the connector routes it to replyAddr.
reply, err := replyRcv.Receive(ctx, nil)
if err != nil {
    log.Fatal(err)
}
fmt.Printf("reply correlation=%v body=%s\n",
    reply.Properties.CorrelationID, reply.GetData())
// receiver.AcceptMessage acknowledges the reply delivery.
replyRcv.AcceptMessage(ctx, reply)

replyRcv.Close(ctx)
snd.Close(ctx)

Security

Authentication

TLS / mTLS

The TLS listener on port 5671 activates only when the top-level Security block is configured. Point clients at amqps://kubemq-host:5671. For mutual TLS, configure the server to request client certificates; the certificate CN then serves as the connection ClientID.

Authorization

With Authorization.Enable = true, the connection's ClientID is checked against the Casbin policy per link:

• Sender link (client → KubeMQ): Write on the resolved channel, checked at ATTACH.
• Receiver link (KubeMQ → client): Read on the resolved channel, checked at ATTACH.
• Anonymous-terminus sender: per-message Write check against properties.to (1024-entry LRU cache, 60 s TTL).
• /responses/<RequestID> reply token: no policy check (connection-scoped).

Minimal TOML configuration

[Connectors.Amqp10]
Enable   = true
Port     = 5672   # shared with [Connectors.Amqp] (0-9-1) via the same listener
TlsPort  = 5671   # active only when [Security] is configured

What Does NOT Migrate / Deviations

Not supported

Behavioral deviations

Verification Smoke Test

This recipe uses the publish and consume snippet above as a copy-pasteable confirmation that the migration is working.

Prerequisites:

1. KubeMQ is running with Connectors.Amqp10.Enable = true.
2. A KubeMQ JWT is available (or Authentication.Enable = false for local testing).

Steps:

// Smoke test — publish one message to /queues/smoke-test, consume it, confirm arrival.
// Run the main() from the Queues snippet above with destination "/queues/smoke-test".
// Expected output: received: <your message body>

1. Run the snippet targeting /queues/smoke-test.
2. Confirm received: ... appears in stdout.
3. If the amqp.Dial call fails, check that CONNECTORS_AMQP10_ENABLE=true is set and port 5672 is reachable.
4. If sender.Send times out, verify the JWT in the SASL PLAIN password field is valid.
5. For Events / Events Store, swap the address prefix and confirm fan-out to multiple receivers.
6. For RPC, run the command snippet and confirm a reply with matching CorrelationID is received within the DefaultRpcTimeoutSeconds window (default: 30 s).

See Also