Skip to content

rabbitmq://

RabbitMQ messaging (AMQP 0-9-1) via rabbitmq/amqp091-go. Exchanges, queues, bindings, publisher confirms, dead-letter routing, retry, TLS, mTLS.

For AMQP 1.0 (Azure Service Bus, ActiveMQ Artemis), use amqp:// — RabbitMQ implements 0-9-1, which is a different wire protocol despite the shared acronym.

URI shape

rabbitmq://<exchangeName>?[options]

<exchangeName> is the AMQP exchange the endpoint binds to. exchangeName="" (empty) targets the default exchange (direct routing by queue name).

Parameters

Connection

Param Type Default Description
exchangeName path — AMQP exchange (path token).
url string amqp://guest:guest@localhost:5672/ Broker URL (amqp:// or amqps://).
username string — PLAIN SASL username. Overrides inline-URL auth.
password string — PLAIN SASL password.
vhost string / AMQP virtual host.
heartbeat duration 10s AMQP heartbeat cadence.
connectionTimeout duration 30s Dial timeout.
connectionName string — Management-UI display name for the connection.
reconnect boolean true Enable automatic reconnect on transport loss.
maxReconnectAttempts integer -1 Max reconnect attempts. -1 = forever.
reconnectTimeWait duration 2s Delay between reconnect attempts.
testConnectionOnStartup boolean false Eager-validate broker connection at component start.

TLS

Param Type Default Description
secure boolean false Force TLS (equivalent to amqps://).
tlsCAFile string — PEM CA bundle.
tlsCertFile string — PEM client cert (mTLS).
tlsKeyFile string — PEM client private key.
tlsInsecureSkipVerify boolean false Skip server cert verification. Insecure.
tlsMinVersion / tlsMaxVersion enum — Pin TLS version.
tlsCipherSuites string — Comma-separated Go cipher-suite names.

Topology

Param Type Default Description
queues string — Comma-separated queue names. Consumer subscribes to each; producer uses first when skipDeclareQueue=false.
routingKey string — Routing key for binding (consumer) / publish (producer). Empty is valid for fanout/headers.
exchangeType enum direct When autoDeclare=true: direct / topic / fanout / headers.
deadLetterExchange string — DLX name. When set, queues declare x-dead-letter-exchange.
deadLetterExchangeType enum direct DLX exchange type.
deadLetterQueue string — DLQ name. autoDeclare creates + binds it to the DLX.
deadLetterRoutingKey string — Routing key for DLX binding.
disableReplyTo boolean false Consumer: ignore inbound reply-to (suppress auto-reply).

Consumer

Param Type Default Description
acknowledgeMode enum auto auto (ack-on-success), manual (processor calls ack), none (broker autoAck).
autoDeclare boolean false Declare exchange + queue + binding on startup.
exclusive boolean false Declare queue as exclusive (single-consumer lock).
noLocal boolean false Don't deliver messages from this same connection.
maximumRetryAttempts integer 0 In-process retry count on processor error. After N retries, falls to DLQ disposition.
retryDelay duration 0 Delay between in-process retries.
rejectAndDontRequeue boolean true On final error, nack without requeue (routes to DLX). False = requeue forever.
concurrentConsumers integer 1 Minimum concurrent consumer goroutines per queue (floor).
maxConcurrentConsumers integer 1 Actual concurrent consumer goroutines per queue.
prefetchCount integer 250 Channel-level basic.qos prefetch count.

Producer

Param Type Default Description
allowNullBody boolean true Allow publishing empty-body messages.
autoDeclareProducer boolean false Declare exchange + queue + binding before first publish.
confirm enum none Publisher confirms mode. correlated / simple wait for broker ack; none is fire-and-forget.
confirmTimeout duration 5s Timeout waiting for publisher confirm.
replyTimeout duration 5s InOut reply timeout via amq.rabbitmq.reply-to.
skipBindQueue boolean false Declare queue without binding to the exchange.
skipDeclareExchange boolean false Skip exchange declaration (use a pre-existing one).
skipDeclareQueue boolean false Skip queue declaration (publish directly to the exchange).
usePublisherConnection boolean false Use a separate pooled connection for publishers (matches Spring AMQP).

Headers

Param Type Description
headerIncludePattern string regex Pass only matching header names. Empty = all.
headerExcludePattern string regex Drop matching header names.

Body shape

Producer

  • In body — published as the message body (BodyAsBytes()).
  • Reads rabbitmq.ROUTING_KEY header (or routingKey URI param) for the publish routing key.
  • rabbitmq.CONTENT_TYPE, rabbitmq.MESSAGE_ID, rabbitmq.CORRELATION_ID headers are mirrored to the AMQP message properties.
  • Other pass-filtered In headers become amqp.Table headers on the message.
  • InOut exchanges use amq.rabbitmq.reply-to for replies; the reply body replaces the In body.

Consumer

For each delivery:

  • In body — message body bytes.
  • In headers:
Header Description
rabbitmq.EXCHANGE Source exchange.
rabbitmq.ROUTING_KEY Source routing key.
rabbitmq.DELIVERY_TAG Delivery tag.
rabbitmq.CONTENT_TYPE Message content-type.
rabbitmq.MESSAGE_ID Message ID.
rabbitmq.CORRELATION_ID Correlation ID.
rabbitmq.REPLY_TO Reply-to (if set).
rabbitmq.REDELIVERED "true" if msg.Redelivered.
rabbitmq.RETRY_COUNT In-process retry count (incremented per attempt).

Plus pass-filtered amqp.Table headers as In headers.

When the inbound delivery has a reply-to and disableReplyTo=false, the processor's output body is auto-published to that queue.

Examples

Subscribe with autoDeclare + DLX:

- id: orders
  from: rabbitmq://orders.ex?url=amqp://broker:5672/&queues=orders.q&routingKey=orders.new&autoDeclare=true&deadLetterExchange=orders.dlx&deadLetterQueue=orders.dlq&maximumRetryAttempts=3&retryDelay=2s
  steps:
    - to: direct://process-order

Publisher with confirms:

- to: rabbitmq://events.ex?url=amqp://broker:5672/&routingKey=event.created&confirm=correlated&confirmTimeout=2s

InOut RPC via amq.rabbitmq.reply-to:

- to: rabbitmq://rpc.ex?url=amqp://broker:5672/&routingKey=rpc.echo&replyTimeout=5s

Manual ack mode:

- from: rabbitmq://work.ex?url=amqp://broker:5672/&queues=work.q&acknowledgeMode=manual
  steps:
    - to: direct://process
    # processor decides ack/nack via headers (see manual-ack docs)

Skip topology, publish to existing exchange:

- to: rabbitmq://existing.ex?url=amqp://broker:5672/&skipDeclareExchange=true&skipDeclareQueue=true&routingKey=foo

Behavior notes

  • AMQP 0-9-1. Despite the acronym, this is not the AMQP 1.0 protocol — RabbitMQ implements 0-9-1, and the wire format / topology semantics differ.
  • Connection split. usePublisherConnection=true opens a separate pooled connection for the producer, mirroring Spring AMQP's recommendation that publishing and consuming shouldn't share a connection (publishers can block consumers via channel flow control).
  • Retry semantics. On processor error, maximumRetryAttempts × retryDelay controls in-process retry. After exhaustion, rejectAndDontRequeue=true (default) nacks without requeue — the message goes to the DLX if configured, otherwise it's discarded. rejectAndDontRequeue=false requeues forever.
  • Header filter is symmetric on both directions; include first, then exclude.
  • autoDeclare is idempotent — repeated declarations of the same exchange/queue/binding with matching properties are silently no-op'd by the broker.
  • confirm=correlated waits per-message; simple waits at flush time. none is fire-and-forget (no broker acknowledgement before Process returns).

Source

  • pkg/component/rabbitmq/rabbitmq.go — Component, Endpoint.
  • pkg/component/rabbitmq/{producer,consumer}.go — channel I/O, retry/DLX logic.
  • pkg/component/rabbitmq/{declare,config,security,connection_pool,headers}.go — topology declaration, config, TLS, pooling, header filter.

See also