mqutils
GitLab ↗

Docs / Backends / AMQP / RabbitMQ

AMQP / RabbitMQ

AMQP 0-9-1 consumer and producer with exchanges, routing keys, dead-letter exchanges, retry queues and channel pooling

go
import "go.digitalxero.dev/mq-amqp/v2"

Index

funcNewAMQPConsumer

go
func NewAMQPConsumer(config *viper.Viper) (types.Consumer, error)

NewAMQPConsumer creates a new AMQP/RabbitMQ consumer with the provided configuration. This function is typically called by mqutils.NewConsumer when it detects an AMQP URL.

Configuration options:

  • url: AMQP connection URL (required, e.g., “amqp://user:pass@host:port/vhost”)
  • queue: Queue name to consume from (required; alias: destination)
  • exchange: Exchange name (optional, for binding)
  • exchange_type: Exchange type - direct, fanout, topic, headers (default: direct)
  • routing_key: Routing key for queue binding (optional)
  • transient: Whether queue is non-durable (default: false)
  • queue_type: Queue type - “classic” or “quorum” (default: classic). Quorum queues must be durable.
  • transient_queue_expires: ms a transient queue that had to be upgraded to durable (RabbitMQ 4.3+) may sit unused before the broker deletes it (default: 0, disabled). Ignored for queues declared durable on purpose.
  • retry_queue_expires: Explicit retry-queue expiry in ms (default: 0, disabled); requires auto_declare and renews the declaration while the consumer is active.
  • max_concurrent_handlers: Maximum active handlers or batches (defaults to the effective message_channel_buffer; must be positive).
  • auto_declare: Auto-create queue/exchange if missing (default: false)
  • auto_reconnect: Automatically reconnect on connection loss (default: false)
  • sni_hostname: TLS ServerName when it differs from the URL host (default: URL host)
  • tls_cert / tls_key: client cert/key PEM paths for mTLS; both required if either is set
  • tls_ca: optional CA PEM; system roots if unset
  • skip_verify: only relaxes certificate verification; does not disable TLS or SNI
  • handler: Name of registered handler function (default: “amqpLogger”)
  • dead_letter_exchange: DLX for failed messages (optional)
  • retry_queue_name: Queue for message retries (optional)
  • retry_queue_ttl: TTL for retry messages in ms (default: 1000)
  • retry_queue_max_retries: Max retry attempts (default: 50; alias: max_retries)
  • message_channel_buffer: Buffer size for message channel (default: 10)

On RabbitMQ 4.3+ with queue_type “quorum” and the in-place retry configuration (retry_queue_name == queue), retries use the broker-native delayed-retry and delivery-limit features and the x-acquired-count header instead of a separate dead-letter retry queue. On older brokers or with retry_queue_name != queue, the classic dead-letter + TTL retry queue is used (counted via x-death).

When retry_queue_name == queue but the broker is pre-4.3 (or the queue is not a quorum queue), the consumer transparently falls back to the classic dead-letter method using a retry queue named “<queue>-retry”. This fallback queue is only auto-created when auto_declare is true; with auto_declare false the caller is responsible for pre-declaring the “<queue>-retry” queue and its bindings, otherwise nacked messages dead-letter to an exchange with no bound queue and are dropped.

  • graceful_shutdown_timeout: Shutdown timeout in seconds (default: 300)
  • enable_graceful_shutdown: Enable graceful shutdown (default: false)
  • batch_size: Number of messages per batch (default: 5)
  • batch_timeout: Batch collection timeout in ms (default: 100)
  • enable_batch_processing: Enable batch message processing (default: false)

Callers that currently read SNI_HOSTNAME, RABBITMQ_SSLCERT, and RABBITMQ_SSLKEY should set sni_hostname, tls_cert, and tls_key on the viper config (or use ConsumerBuilder.WithTLSServerName / WithTLSClientCert). This package does not read those environment variables.

Returns an error if configuration validation fails or transport creation fails.

funcNewAMQPProducer

go
func NewAMQPProducer(config *viper.Viper) (types.Producer, error)

NewAMQPProducer creates a new AMQP/RabbitMQ producer with the provided configuration. This function is typically called by mqutils.NewProducer when it detects an AMQP URL.

Configuration options:

  • url: AMQP connection URL (required, e.g., “amqp://user:pass@host:port/vhost”)
  • exchange: Exchange name for publishing (required; alias: destination)
  • exchange_type: Exchange type - direct, fanout, topic, headers (default: topic)
  • routing_key: Default routing key for published messages (optional)
  • mandatory: Publish with the mandatory flag so unroutable messages are returned by the broker instead of silently dropped (default: true)
  • durable: Whether the declared exchange is durable (default: true)
  • auto_delete: Whether the declared exchange is deleted when unused (default: false)
  • sni_hostname: TLS ServerName when it differs from the URL host (default: URL host)
  • tls_cert / tls_key: client cert/key PEM paths for mTLS; both required if either is set
  • tls_ca: optional CA PEM; system roots if unset
  • skip_verify: only relaxes certificate verification; does not disable TLS or SNI
  • priority: Default message priority 0-9 (default: 0)
  • delivery_mode: Message persistence - 1 (non-persistent), 2 (persistent) (default: 2)
  • channel_pool_size: Max pooled channels for publishing (default: 20). The channel pool is shared process-wide per connection URL + TLS identity and is sized by whichever producer or consumer connects first; if a pool already exists for the URL and a different non-default size is requested, a warning is logged and the existing pool is reused as-is.
  • publisher_confirms: Wait for the broker to confirm every publish (default: false). A broker nack returns types.ErrPublishNacked. While the broker has the connection BLOCKED (memory/disk alarm), confirmed publishes are rejected immediately with types.ErrConnectionBlocked — the caller decides how to handle it.
  • confirm_timeout: Max wait for a broker confirmation as a duration string (default: “1s”); the caller context still wins when earlier.
  • publish_queue_size: Capacity of the internal queue that fire-and-forget publishers use while the connection is blocked (default: 1000). Queued publishes are flushed in order once the broker unblocks; when the queue is full, publishes return types.ErrPublishQueueFull. Not used when publisher_confirms is enabled.

Callers that currently read SNI_HOSTNAME, RABBITMQ_SSLCERT, and RABBITMQ_SSLKEY should set sni_hostname, tls_cert, and tls_key on the viper config (or use ProducerBuilder.WithTLSServerName / WithTLSClientCert). This package does not read those environment variables.

Returns an error if configuration validation fails or transport creation fails. The producer must be started with Start() before publishing messages.

funcNewAMQPTransport

go
func NewAMQPTransport(retry bool, prefetch int) types.Transport

NewAMQPTransport creates a new AMQP transport implementation. The transport provides low-level AMQP operations including connection management, channel pooling, message publishing, and consumption.

Parameters:

  • retry: Whether to retry failed operations (used for retry queue handling)
  • prefetch: Number of messages to prefetch for fair dispatching (0 = unlimited)

The transport uses connection pooling to efficiently manage AMQP resources and provides separate channels for consumer operations and publishing to avoid conflicts and improve performance.

Transports created through this constructor use the historical defaults: mandatory publishing, durable exchanges that are not auto-deleted, and the default publisher-channel pool size. Callers that need to configure those (the producer) use newAMQPTransport directly.

funcNewMessageBuilder

go
func NewMessageBuilder() types.MessageBuilder

NewMessageBuilder creates a new AMQP message builder. The builder provides a fluent interface for constructing AMQP messages with all supported properties and attributes.

Example:

go
msg := amqp.NewMessageBuilder().
    WithBody([]byte("Hello, World!")).
    WithContentType("text/plain").
    WithCorrelationId("req-123").
    WithHeaders(map[string]interface{}{"source": "service-a"}).
    WithDeliveryMode(2). // Persistent
    Build()

Generated by gomarkdoc

↑ ↓ move↵ open/ opens search anywhere