amqp://¶

AMQP 1.0 messaging via github.com/Azure/go-amqp. Sender/Receiver links, dynamic reply-to, settled vs unsettled delivery, SASL PLAIN/ANONYMOUS/EXTERNAL, TLS + mTLS. JMS-bean parameters (connection factories, transactions) are intentionally not part of the surface.

For AMQP 0-9-1 (RabbitMQ wire protocol), use rabbitmq:// — same dial syntax, different on-wire protocol.

URI shape¶

amqp://<address>?[options]

<address> is the queue / topic name (or any AMQP 1.0 source/target address). Use destinationType=queue|topic|temp-queue|temp-topic to hint the broker convention.

Parameters¶

Connection¶

TLS¶

Address¶

Common¶

Consumer¶

In CLIENT_ACKNOWLEDGE mode the processor controls the disposition via the amqp.SETTLE header on the outbound exchange (accept / release / reject). A missing/unknown value defaults to release (forces redelivery) — matching JMS CLIENT_ACKNOWLEDGE semantics.

Producer¶

Headers¶

Body shape¶

Producer¶

• In body — sent as the AMQP message data (Data[0] from BodyAsBytes()).
• In headers — pass-filtered → application-properties.
• InOut exchanges open a per-link reply receiver (dynamic source); reply body replaces the In body.

Consumer¶

For each received message:

• In body — message data bytes.
• In headers:

Plus pass-filtered application-properties as In headers.

Headers¶

The constants are exported in code as:

HeaderSettle values: accept / release / reject.

Examples¶

Consume from an AMQP queue:

- id: orders
  from: amqp://orders?url=amqp://guest:guest@broker:5672/&prefetchCount=100&maxConcurrentConsumers=4
  steps:
    - to: direct://process-order

Send with persistent delivery + priority:

- setHeader:
    name: amqp.PRIORITY
    constant: "9"
- to: amqp://critical?url=amqp://broker:5672/&deliveryMode=PERSISTENT

InOut request/reply (dynamic reply-to):

- to: amqp://inventory.lookup?url=amqp://broker:5672/&requestTimeout=2s

TLS + mTLS via SASL EXTERNAL:

- to: amqp://secure?url=amqps://broker.example.com:5671/&sasl=EXTERNAL&tlsCertFile=/etc/ssl/client.crt&tlsKeyFile=/etc/ssl/client.key&tlsCAFile=/etc/ssl/ca.pem

CLIENT_ACKNOWLEDGE with selective dispositions:

- from: amqp://work?url=amqp://broker:5672/&acknowledgementModeName=CLIENT_ACKNOWLEDGE
  steps:
    - choice:
        when:
          - starlang: 'body == "poison"'
            steps:
              - setHeader:
                  name: amqp.SETTLE
                  constant: reject       # dead-letter
        otherwise:
          steps:
            - to: direct://process
            - setHeader:
                name: amqp.SETTLE
                constant: accept

Behavior notes¶

• AMQP 1.0 only. For RabbitMQ-style AMQP 0-9-1, use rabbitmq://.
• Connection pool. Endpoints with identical connection config share a connection via connection_pool.go.
• maxConcurrentConsumers. Auto-promoted to at least concurrentConsumers. Each goroutine maintains its own Receiver link.
• CLIENT_ACKNOWLEDGE default disposition is release. Matches JMS — forgetting to set amqp.SETTLE causes the broker to redeliver.
• Header filter is symmetric. The same regex governs both directions; the include is applied first, then the exclude.
• Reconnect. A Reconnector resets the entire link on transport failure. recoveryInterval sets the delay.

Source¶

• pkg/component/amqp/amqp.go — Component, Endpoint.
• pkg/component/amqp/{producer,consumer}.go — link I/O, settle handling.
• pkg/component/amqp/{config,security,connection_pool,headers,address}.go — config parsing, TLS+SASL, pooling, header filter, address types.

See also¶

• AMQP 0-9-1 (RabbitMQ): rabbitmq://
• Other brokers: kafka://, nats://, mqtt://