diff --git a/CHANGELOG.md b/CHANGELOG.md index 713b3f4..723bd94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,26 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.7.0] - 2026-07-05 + +### Added + +- **One-call dead-letter queue setup.** `ConsumerConfig.WithDeadLetterQueue` + (with the new `DeadLetterConfig` type and `DefaultDeadLetterConfig` helper) + declares a dead-letter exchange, a dead-letter queue, the binding between them, + and wires the work queue's `x-dead-letter-exchange` — previously four separate + hand-written declarations. Like the rest of the consumer topology (v0.5.0), it + is re-applied on every channel setup, so it survives reconnects and broker + restarts. `DefaultDeadLetterConfig("work")` derives `work.dlx` / `work.dlq` + (durable fanout); tune with `WithExchange`/`WithQueue`/`WithRoutingKey`/ + `WithDurable`/`WithQuorum`/`WithMaxLength`/`WithMessageTTL`. +- `Consumer.DeadLetterQueueName()` returns the configured DLQ name, so a second + consumer can read dead-lettered messages without re-deriving it. + + This pairs with the v0.6.0 default (`RequeueOnError=false`): a failed handler + now rejects, and with a dead-letter queue configured the message is captured + instead of discarded. + ## [0.6.0] - 2026-07-05 ### Changed diff --git a/README.md b/README.md index 0d6e8fa..6b9a94d 100644 --- a/README.md +++ b/README.md @@ -247,6 +247,33 @@ automatically and consumption resumes. `WithBinding` also works for server-named queues (empty queue name), which get a fresh name on each reconnect. The bound exchange must already exist when the consumer is created. +### Dead-Letter Queues + +`WithDeadLetterQueue` sets up a work queue's dead-letter topology in one call — +it declares the dead-letter exchange, the dead-letter queue, the binding between +them, and wires the work queue to dead-letter into it. Like the rest of the +topology, it is re-applied on every reconnect. Combined with the default +`RequeueOnError: false`, a failed handler's message is captured on the DLQ +instead of being requeued or discarded: + +```go +consConfig := rabbitmq.DefaultConsumerConfig(). + WithQueueConfig(rabbitmq.DefaultQueueConfig("orders")). + WithDeadLetterQueue(rabbitmq.DefaultDeadLetterConfig("orders")) // orders.dlx / orders.dlq + +consumer, err := rabbitmq.NewConsumer(conn, consConfig) +// ... consume "orders"; failures are dead-lettered automatically. + +// Read dead-lettered messages like any other queue: +dlq, _ := rabbitmq.NewConsumer(conn, + rabbitmq.DefaultConsumerConfig().WithQueue(consumer.DeadLetterQueueName())) +``` + +`DefaultDeadLetterConfig("orders")` derives a durable fanout `orders.dlx` and a +durable `orders.dlq`; tune names, durability, quorum, max-length, or a TTL with +the `With*` builders on `DeadLetterConfig`. The work queue must have a name (it +carries the dead-letter wiring). + ### Concurrent Consumers Process messages in parallel with multiple worker goroutines: @@ -302,11 +329,11 @@ consConfig := rabbitmq.DefaultConsumerConfig(). #### Built-in Middleware -| Middleware | Description | -|---|---| -| `LoggingMiddleware(logger)` | Logs message processing with duration | -| `RecoveryMiddleware(onPanic)` | Recovers from panics in handlers | -| `RetryMiddleware(maxRetries, delay)` | Retries failed message processing | +| Middleware | Description | +| ------------------------------------ | ------------------------------------- | +| `LoggingMiddleware(logger)` | Logs message processing with duration | +| `RecoveryMiddleware(onPanic)` | Recovers from panics in handlers | +| `RetryMiddleware(maxRetries, delay)` | Retries failed message processing | #### Custom Middleware diff --git a/consumer.go b/consumer.go index ecf8853..e786252 100644 --- a/consumer.go +++ b/consumer.go @@ -68,6 +68,11 @@ type ConsumerConfig struct { // Bindings are applied to the consumed queue on every channel setup, // initially and after each reconnect. Bindings []BindingConfig + + // DeadLetter, when set, declares a dead-letter exchange, queue, and binding + // and wires the work queue to dead-letter into it — on every channel setup, + // initially and after each reconnect. Set it with WithDeadLetterQueue. + DeadLetter *DeadLetterConfig } // DefaultConsumerConfig returns a default consumer configuration. @@ -175,6 +180,45 @@ func (c ConsumerConfig) WithBinding(exchange, routingKey string, args map[string return c } +// WithDeadLetterQueue returns a new config that declares a dead-letter exchange, +// a dead-letter queue, and their binding, and wires the work queue to +// dead-letter into that exchange — all on every channel setup, so the topology +// survives reconnects and broker restarts. It is the one-call equivalent of +// hand-declaring the DLX, the DLQ, the binding, and the work queue's +// x-dead-letter-exchange argument. +// +// The work queue must be named (via WithQueue or WithQueueConfig): the DLX +// wiring requires a consumer-declared queue, so NewConsumer rejects a +// dead-letter config with an anonymous work queue (returning ErrInvalidConfig) +// rather than declaring an orphan server-named queue. Consume the dead-letter +// queue like any other queue using its name (see DeadLetterQueueName or dl.Queue). +func (c ConsumerConfig) WithDeadLetterQueue(dl DeadLetterConfig) ConsumerConfig { + c.DeadLetter = &dl + + // The work queue must carry x-dead-letter-exchange, which requires it to be + // consumer-declared. Synthesize a QueueConfig from the resolved queue name. + // With no name, leave QueueConfig unset so NewConsumer rejects the config + // instead of declaring an orphan (durable, empty-name) server-named queue. + name := c.Queue + if c.QueueConfig != nil { + name = c.QueueConfig.Name + } + if name == "" { + return c + } + + qc := DefaultQueueConfig(name) + if c.QueueConfig != nil { + qc = *c.QueueConfig + } + qc.DeadLetterExchange = dl.Exchange + qc.DeadLetterRoutingKey = dl.RoutingKey + c.QueueConfig = &qc + c.Queue = qc.Name + + return c +} + // Consumer consumes messages from RabbitMQ. type Consumer struct { conn *Connection @@ -212,6 +256,12 @@ func NewConsumer(conn *Connection, config ConsumerConfig) (*Consumer, error) { queueName = config.QueueConfig.Name } + // Dead-lettering requires a named work queue to carry x-dead-letter-exchange; + // reject an anonymous one rather than declaring orphan queues on each setup. + if config.DeadLetter != nil && queueName == "" { + return nil, fmt.Errorf("%w: WithDeadLetterQueue requires a named work queue", ErrInvalidConfig) + } + c := &Consumer{ conn: conn, config: config, @@ -277,6 +327,12 @@ func (c *Consumer) setupChannel() error { // applyTopology declares the configured queue and applies the configured // bindings on the given channel, returning the resolved queue name. func (c *Consumer) applyTopology(ch *Channel) (string, error) { + // Declare the dead-letter topology first so the work queue's + // x-dead-letter-exchange target exists before the work queue is declared. + if err := c.applyDeadLetter(ch); err != nil { + return "", err + } + queueName := c.config.Queue switch { @@ -307,6 +363,34 @@ func (c *Consumer) applyTopology(ch *Channel) (string, error) { return queueName, nil } +// applyDeadLetter idempotently declares the configured dead-letter exchange, +// dead-letter queue, and the binding between them. It is a no-op when no +// DeadLetterConfig is set. +func (c *Consumer) applyDeadLetter(ch *Channel) error { + if c.config.DeadLetter == nil { + return nil + } + dl := c.config.DeadLetter + + kind := dl.ExchangeType + if kind == "" { + kind = ExchangeFanout + } + if err := ch.ch.ExchangeDeclare(dl.Exchange, string(kind), dl.Durable, false /*autoDelete*/, false /*internal*/, false /*noWait*/, nil); err != nil { + return fmt.Errorf("declare dead-letter exchange %q: %w", dl.Exchange, err) + } + + if _, err := ch.ch.QueueDeclare(dl.Queue, dl.Durable, false /*autoDelete*/, false /*exclusive*/, false /*noWait*/, amqp.Table(dl.buildArgs())); err != nil { + return fmt.Errorf("declare dead-letter queue %q: %w", dl.Queue, err) + } + + if err := ch.ch.QueueBind(dl.Queue, dl.RoutingKey, dl.Exchange, false /*noWait*/, nil); err != nil { + return fmt.Errorf("bind dead-letter queue %q to exchange %q: %w", dl.Queue, dl.Exchange, err) + } + + return nil +} + // QueueName returns the queue this consumer reads from. When NewConsumer was // given an empty queue name, this is the server-assigned name (available after // construction). @@ -316,6 +400,16 @@ func (c *Consumer) QueueName() string { return c.queue } +// DeadLetterQueueName returns the name of the dead-letter queue configured with +// WithDeadLetterQueue, or "" if none was configured. Use it to consume or +// inspect dead-lettered messages. +func (c *Consumer) DeadLetterQueueName() string { + if c.config.DeadLetter == nil { + return "" + } + return c.config.DeadLetter.Queue +} + // Start starts consuming messages and returns a delivery channel. // The delivery channel is automatically re-established on reconnection. func (c *Consumer) Start(ctx context.Context) (<-chan *Delivery, error) { @@ -717,6 +811,104 @@ type BindingConfig struct { Args map[string]any } +// DeadLetterConfig describes the dead-letter topology declared alongside a +// consumer's work queue (see ConsumerConfig.WithDeadLetterQueue): a dead-letter +// exchange, a dead-letter queue, and the binding between them. It is declared +// idempotently on every channel setup, so it survives reconnects and broker +// restarts. +type DeadLetterConfig struct { + // Exchange is the dead-letter exchange (DLX) name. + Exchange string + + // ExchangeType is the DLX type (default: fanout). + ExchangeType ExchangeType + + // Queue is the dead-letter queue (DLQ) name. + Queue string + + // RoutingKey routes the work queue's rejected messages to the DLX and binds + // the DLQ to it (default: "", which suits a fanout DLX). + RoutingKey string + + // Durable sets the DLX and DLQ durability (default: true). + Durable bool + + // Quorum declares the DLQ as a quorum queue. + Quorum bool + + // MaxLength caps the number of messages in the DLQ (0 = unbounded). + MaxLength int + + // MessageTTL expires DLQ entries after the given duration (0 = never). + MessageTTL time.Duration +} + +// DefaultDeadLetterConfig returns a dead-letter config for the given work queue, +// deriving ".dlx" and ".dlq" names, a durable fanout DLX, +// and a durable DLQ. +func DefaultDeadLetterConfig(workQueue string) DeadLetterConfig { + return DeadLetterConfig{ + Exchange: workQueue + ".dlx", + ExchangeType: ExchangeFanout, + Queue: workQueue + ".dlq", + Durable: true, + } +} + +// WithExchange returns a new config with the specified DLX name and type. +func (c DeadLetterConfig) WithExchange(name string, kind ExchangeType) DeadLetterConfig { + c.Exchange = name + c.ExchangeType = kind + return c +} + +// WithQueue returns a new config with the specified DLQ name. +func (c DeadLetterConfig) WithQueue(name string) DeadLetterConfig { + c.Queue = name + return c +} + +// WithRoutingKey returns a new config with the specified dead-letter routing key. +func (c DeadLetterConfig) WithRoutingKey(key string) DeadLetterConfig { + c.RoutingKey = key + return c +} + +// WithDurable returns a new config with the specified durability for the DLX and DLQ. +func (c DeadLetterConfig) WithDurable(durable bool) DeadLetterConfig { + c.Durable = durable + return c +} + +// WithQuorum returns a new config that declares the DLQ as a durable quorum queue. +func (c DeadLetterConfig) WithQuorum() DeadLetterConfig { + c.Quorum = true + c.Durable = true // quorum queues must be durable + return c +} + +// WithMaxLength returns a new config that caps the DLQ length. +func (c DeadLetterConfig) WithMaxLength(maxLength int) DeadLetterConfig { + c.MaxLength = maxLength + return c +} + +// WithMessageTTL returns a new config that expires DLQ entries after ttl. +func (c DeadLetterConfig) WithMessageTTL(ttl time.Duration) DeadLetterConfig { + c.MessageTTL = ttl + return c +} + +// buildArgs builds the DLQ declaration arguments. It reuses QueueConfig.buildArgs +// so the argument keys stay defined in one place. +func (c DeadLetterConfig) buildArgs() map[string]any { + return QueueConfig{ + Quorum: c.Quorum, + MaxLength: c.MaxLength, + MessageTTL: c.MessageTTL, + }.buildArgs() +} + // QueueConfig holds queue declaration configuration. type QueueConfig struct { // Name is the queue name. diff --git a/consumer_test.go b/consumer_test.go index 3c083d9..0252cca 100644 --- a/consumer_test.go +++ b/consumer_test.go @@ -434,3 +434,111 @@ func TestConsumerConfigWithBindingCopySemantics(t *testing.T) { t.Errorf("expected b's second binding to be b, got %s", b.Bindings[1].RoutingKey) } } + +func TestDefaultDeadLetterConfig(t *testing.T) { + dl := DefaultDeadLetterConfig("work") + if dl.Exchange != "work.dlx" { + t.Errorf("Exchange = %q, want work.dlx", dl.Exchange) + } + if dl.Queue != "work.dlq" { + t.Errorf("Queue = %q, want work.dlq", dl.Queue) + } + if dl.ExchangeType != ExchangeFanout { + t.Errorf("ExchangeType = %q, want fanout", dl.ExchangeType) + } + if !dl.Durable { + t.Error("expected Durable true") + } +} + +func TestDeadLetterConfigBuilders(t *testing.T) { + dl := DefaultDeadLetterConfig("work"). + WithExchange("custom.dlx", ExchangeDirect). + WithQueue("custom.dlq"). + WithRoutingKey("dead"). + WithMaxLength(1000). + WithMessageTTL(time.Hour) + + if dl.Exchange != "custom.dlx" || dl.ExchangeType != ExchangeDirect { + t.Errorf("unexpected exchange: %q/%q", dl.Exchange, dl.ExchangeType) + } + if dl.Queue != "custom.dlq" || dl.RoutingKey != "dead" { + t.Errorf("unexpected queue/key: %q/%q", dl.Queue, dl.RoutingKey) + } + + args := dl.buildArgs() + if args["x-max-length"] != 1000 { + t.Errorf("x-max-length = %v, want 1000", args["x-max-length"]) + } + if args["x-message-ttl"] != time.Hour.Milliseconds() { + t.Errorf("x-message-ttl = %v, want %d", args["x-message-ttl"], time.Hour.Milliseconds()) + } + + if q := DefaultDeadLetterConfig("w").WithQuorum(); !q.Quorum || !q.Durable { + t.Error("WithQuorum should set Quorum and force Durable") + } + if qa := DefaultDeadLetterConfig("w").WithQuorum().buildArgs(); qa["x-queue-type"] != "quorum" { + t.Errorf("x-queue-type = %v, want quorum", qa["x-queue-type"]) + } +} + +func TestWithDeadLetterQueue_SynthesizesQueueConfig(t *testing.T) { + // No QueueConfig set: WithDeadLetterQueue synthesizes one from the queue name + // and stamps the DLX wiring onto it. + c := DefaultConsumerConfig(). + WithQueue("orders"). + WithDeadLetterQueue(DefaultDeadLetterConfig("orders")) + + if c.DeadLetter == nil { + t.Fatal("expected DeadLetter to be set") + } + if c.QueueConfig == nil { + t.Fatal("expected a synthesized QueueConfig") + } + if c.QueueConfig.Name != "orders" { + t.Errorf("work queue name = %q, want orders", c.QueueConfig.Name) + } + if c.QueueConfig.DeadLetterExchange != "orders.dlx" { + t.Errorf("DeadLetterExchange = %q, want orders.dlx", c.QueueConfig.DeadLetterExchange) + } + // The work queue's args must carry x-dead-letter-exchange. + if c.QueueConfig.buildArgs()["x-dead-letter-exchange"] != "orders.dlx" { + t.Error("work queue args missing x-dead-letter-exchange") + } +} + +func TestWithDeadLetterQueue_PreservesExistingQueueConfig(t *testing.T) { + // An existing QueueConfig keeps its other fields; only DLX wiring is added. + c := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig("orders").WithQuorum().WithMaxLength(500)). + WithDeadLetterQueue(DefaultDeadLetterConfig("orders").WithRoutingKey("dead")) + + if !c.QueueConfig.Quorum { + t.Error("existing Quorum setting was lost") + } + if c.QueueConfig.MaxLength != 500 { + t.Errorf("existing MaxLength lost: got %d", c.QueueConfig.MaxLength) + } + if c.QueueConfig.DeadLetterExchange != "orders.dlx" || c.QueueConfig.DeadLetterRoutingKey != "dead" { + t.Errorf("DLX wiring not applied: %q/%q", c.QueueConfig.DeadLetterExchange, c.QueueConfig.DeadLetterRoutingKey) + } +} + +func TestWithDeadLetterQueue_StoresCopy(t *testing.T) { + dl := DefaultDeadLetterConfig("orders") + c := DefaultConsumerConfig().WithQueue("orders").WithDeadLetterQueue(dl) + dl.Queue = "mutated" + if c.DeadLetter.Queue != "orders.dlq" { + t.Errorf("stored DeadLetter should be a copy; got Queue=%q", c.DeadLetter.Queue) + } +} + +func TestNewConsumer_DeadLetterRequiresNamedQueue(t *testing.T) { + // A dead-letter config without a named work queue must be rejected, not + // silently turned into an orphan server-named queue. + _, err := NewConsumer(&Connection{}, DefaultConsumerConfig(). + WithDeadLetterQueue(DefaultDeadLetterConfig("orders"))) + if !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for anonymous work queue, got %v", err) + } +} diff --git a/integration_test.go b/integration_test.go index 5160cab..c0f0e9f 100644 --- a/integration_test.go +++ b/integration_test.go @@ -2185,3 +2185,130 @@ func TestIntegration_ErrRequeueRedelivers(t *testing.T) { t.Errorf("handler ran %d times, want >= 2 (requeued then acked)", n) } } + +// --- Dead-letter queue helper --- + +// TestIntegration_DeadLetterQueueHelper verifies the one-call WithDeadLetterQueue +// setup: a failed handler (default RequeueOnError=false) rejects the message, and +// it is dead-lettered to the auto-declared DLQ, which a second consumer receives. +func TestIntegration_DeadLetterQueueHelper(t *testing.T) { + conn := integrationConn(t) + work := uniqueQueue(t) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + // Consumer with a one-call DLQ setup. Work queue exclusive to satisfy strict + // brokers (transient non-exclusive queues are rejected on 4.x). + cfg := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(work).WithDurable(false).WithAutoDelete(true).WithExclusive(true)). + WithDeadLetterQueue(DefaultDeadLetterConfig(work)) + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + // The DLX/DLQ are durable; clean them up explicitly. + t.Cleanup(func() { deleteExchange(t, conn, work+".dlx") }) + t.Cleanup(func() { deleteQueue(t, conn, work+".dlq") }) + + if consumer.DeadLetterQueueName() != work+".dlq" { + t.Errorf("DeadLetterQueueName() = %q, want %s.dlq", consumer.DeadLetterQueueName(), work) + } + + // Second consumer reading the auto-declared DLQ. + dlqConsumer, err := NewConsumer(conn, DefaultConsumerConfig().WithQueue(consumer.DeadLetterQueueName())) + if err != nil { + t.Fatalf("failed to create dlq consumer: %v", err) + } + t.Cleanup(func() { dlqConsumer.Close() }) + dlqCh, err := dlqConsumer.Start(ctx) + if err != nil { + t.Fatalf("failed to start dlq consumer: %v", err) + } + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(work)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + go func() { + _ = consumer.Consume(ctx, func(_ context.Context, _ *Delivery) error { + return errors.New("always fails") + }) + }() + + if err := pub.Publish(ctx, NewTextMessage("poison")); err != nil { + t.Fatalf("publish: %v", err) + } + + recvDelivery(t, ctx, dlqCh, "poison") +} + +// TestIntegration_DeadLetterTopologyRestoredAfterReconnect verifies the DLX, DLQ, +// binding, and work-queue dead-letter wiring are re-declared after a connection +// loss, so dead-lettering still works once the consumer recovers. +func TestIntegration_DeadLetterTopologyRestoredAfterReconnect(t *testing.T) { + conn := integrationConn(t) + work := uniqueQueue(t) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + cfg := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(work).WithDurable(false).WithAutoDelete(true).WithExclusive(true)). + WithDeadLetterQueue(DefaultDeadLetterConfig(work)) + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + t.Cleanup(func() { deleteExchange(t, conn, work+".dlx") }) + t.Cleanup(func() { deleteQueue(t, conn, work+".dlq") }) + + go func() { + _ = consumer.Consume(ctx, func(_ context.Context, _ *Delivery) error { + return errors.New("always fails") + }) + }() + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(work)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + // DLQ consumer, created before the drop so it recovers via its own reconnect + // logic (the durable DLQ survives the connection loss). + dlqConsumer, err := NewConsumer(conn, DefaultConsumerConfig().WithQueue(work+".dlq")) + if err != nil { + t.Fatalf("failed to create dlq consumer: %v", err) + } + t.Cleanup(func() { dlqConsumer.Close() }) + dlqCh, err := dlqConsumer.Start(ctx) + if err != nil { + t.Fatalf("failed to start dlq consumer: %v", err) + } + + // Sanity: dead-lettering works before the drop. A single message so the DLQ + // is drained again before the post-reconnect check (avoids a leftover + // "before" being delivered during the "after" phase). + if err := pub.Publish(ctx, NewTextMessage("before")); err != nil { + t.Fatalf("publish before: %v", err) + } + recvDelivery(t, ctx, dlqCh, "before") + + // Force a reconnect: the exclusive work queue is dropped by the broker and + // must be re-declared along with its dead-letter wiring. + conn.mu.RLock() + underlying := conn.conn + conn.mu.RUnlock() + if err := underlying.Close(); err != nil { + t.Fatalf("close underlying connection: %v", err) + } + + // After recovery, dead-lettering must still work — proving the DLX, DLQ, + // binding, and work-queue wiring were all re-declared. + publishUntilDelivered(t, ctx, pub, dlqCh, "", work, "after") +}