diff --git a/README.md b/README.md index 881557f..4707c97 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,10 @@ testing can enable confirmed delivery and call `ReadMessagesToDeadletter`. The method preserves the original body and acknowledges it only after the broker confirms the dead-letter envelope. +RabbitMQ also offers an optional `RabbitOptions.DeadLetterObserver` callback for +payload-free final DLQ publish outcomes, including failures and confirmation +status. See [dead-letter publish outcomes](README.rabbitmq.md#dead-letter-publish-outcomes). + ### Pipeline Actions Message handlers return a `models.PipelineAction`. The queue client maps that diff --git a/README.rabbitmq.md b/README.rabbitmq.md index c01c71c..8d6f1de 100644 --- a/README.rabbitmq.md +++ b/README.rabbitmq.md @@ -64,6 +64,7 @@ methods or channels is required. | `PrefetchCount` | No | Maximum unacknowledged deliveries; defaults to `5` | | `MaxRetries` | No | Retry limit before dead-lettering; defaults to `10` | | `ConfirmedDelivery` | No | Provisions the confirmed producer used by explicit confirmed methods; defaults to `false` | +| `DeadLetterObserver` | No | Optional `func(DeadLetterPublishEvent)` observing final DLQ publish outcomes | | `Exchange` | No | Compatibility field; the default exchange is used | | `Uri` | No | Compatibility field; connection uses host and credentials | | `TLS` | No | Enables an `amqps://` connection | @@ -147,6 +148,48 @@ rabbit.SetReturnHandler(func(returned amqp.Return) { }) ``` +### Dead-letter publish outcomes + +Set `RabbitOptions.DeadLetterObserver` or use `SetDeadLetterObserver` before +creating the client: + +```go +options := queue.NewRabbitOptions(). + SetDeadLetterObserver(func(event queue.DeadLetterPublishEvent) { + // Send these fields to your application's structured logger or metrics. + // Do not treat Err == nil alone as proof of durable delivery. + }) +``` + +The callback runs once per logical publish to the configured DLQ, **after** its +final result (including reconnect), not when a handler returns `PipelineError`. +It covers malformed messages, handler errors, failed forwarding, exhausted +retries, forced `ReadMessagesToDeadletter` transfers, `AddToDeadletter`, direct +`Publish`/`PublishConfirmed`, delayed DLQ publishes, and `PublishDeadLetter`. +Inspection, scanning, replay, discard, and restoration of retained messages do +not emit new-entry events. Validation failures before publishing do not emit. + +`Source`, `Reason`, and `Attempts` come from the envelope. `Destination` is the +actual configured DLQ, not legacy envelope destination metadata that may identify +a replay queue. `TraceID` and `MediaFileName` are extracted only from the inner +JSON's `traceId` and `payload.key` (falling back to top-level `fileName`). +Malformed inner bodies still produce outcomes, with empty correlation fields. +No payload or arbitrary attributes are exposed, and the observer adds no logger +or tracing dependency. + +`Confirmed` is true only when a broker-confirmed publish succeeds (`Err == nil`). +An unconfirmed publish can return nil before a mandatory return arrives; its +event remains `Confirmed: false`. Failed publishes retain their error even if +subsequent disaster recovery succeeds. Observation precedes source settlement, +so success does not imply the source acknowledgement succeeded. Redelivery can +produce another logical publish and event. + +Callbacks run synchronously outside publishing locks, including on the delayed +publisher goroutine. They may run concurrently: handlers must be concurrency-safe, +return promptly, and not panic. A slow callback delays settlement; avoid recursive +DLQ publishing from the callback. Do not mutate options after starting the client. +When unset, no envelope or correlation decoding is performed. + ## TLS Prefixing `Host` with `amqps://` automatically enables TLS. TLS can also be diff --git a/pkg/queue/deadletter_observer.go b/pkg/queue/deadletter_observer.go new file mode 100644 index 0000000..93621b6 --- /dev/null +++ b/pkg/queue/deadletter_observer.go @@ -0,0 +1,65 @@ +package queue + +import "encoding/json" + +// DeadLetterPublishEvent reports one logical RabbitMQ DLQ publish after its final +// outcome, including any reconnect attempt, but before delivery settlement or +// disaster recovery. It contains no message body. +type DeadLetterPublishEvent struct { + Source string + Destination string // Actual configured DLQ, not an envelope's replay destination. + Reason DeadLetterReason + Attempts int + TraceID string + MediaFileName string + Confirmed bool // True only for a broker-confirmed publish with Err == nil. + Err error +} + +func (r *RabbitMQ) publishAndObserveDeadLetter(destination string, body []byte, confirmed bool, publish func() error) error { + observer := r.options.DeadLetterObserver + if observer == nil || destination != r.options.DeadletterQueue { + return publish() + } + + err := publish() + event := DeadLetterPublishEvent{ + Source: r.options.ConsumerQueue, + Destination: destination, + Reason: DeadLetterReasonUnspecified, + Confirmed: confirmed && err == nil, + Err: err, + } + // Project only routing metadata and correlation fields, never the complete + // application model (which may contain credentials or other sensitive data). + var envelope struct { + Schema string `json:"schema"` + Payload []byte `json:"payload"` + DeadLetter struct { + Source string `json:"source"` + Reason DeadLetterReason `json:"reason"` + Attempts int `json:"attempts"` + } `json:"deadLetter"` + } + if json.Unmarshal(body, &envelope) == nil && envelope.Schema == DeadLetterSchemaV1 { + event.Source = envelope.DeadLetter.Source + event.Reason = envelope.DeadLetter.Reason + event.Attempts = envelope.DeadLetter.Attempts + var correlation struct { + TraceID string `json:"traceId"` + FileName string `json:"fileName"` + Payload struct { + Key string `json:"key"` + } `json:"payload"` + } + if json.Unmarshal(envelope.Payload, &correlation) == nil { + event.TraceID = correlation.TraceID + event.MediaFileName = correlation.Payload.Key + if event.MediaFileName == "" { + event.MediaFileName = correlation.FileName + } + } + } + observer(event) + return err +} diff --git a/pkg/queue/deadletter_observer_test.go b/pkg/queue/deadletter_observer_test.go new file mode 100644 index 0000000..95ddfdb --- /dev/null +++ b/pkg/queue/deadletter_observer_test.go @@ -0,0 +1,347 @@ +package queue + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + amqp "github.com/rabbitmq/amqp091-go" +) + +func TestRabbitDeadLetterObserverOutcomes(t *testing.T) { + failed := errors.New("publish failed") + closed := errors.New("channel/connection is not open") + for _, test := range []struct { + name string + confirmed bool + firstErr error + finalErr error + connectErr error + reconnectErr error + wantAttempts int + wantErr error + }{ + {name: "confirmed", confirmed: true, wantAttempts: 1}, + {name: "unconfirmed", wantAttempts: 1}, + {name: "failure", confirmed: true, firstErr: failed, wantAttempts: 1, wantErr: failed}, + {name: "connection failure", confirmed: true, connectErr: failed, wantErr: failed}, + {name: "reconnect success", confirmed: true, firstErr: closed, wantAttempts: 2}, + {name: "legacy reconnect success", firstErr: closed, wantAttempts: 2}, + {name: "reconnect publish failure", confirmed: true, firstErr: closed, finalErr: failed, wantAttempts: 2, wantErr: failed}, + {name: "reconnect failure preserves original error", confirmed: true, firstErr: closed, reconnectErr: failed, wantAttempts: 1, wantErr: closed}, + } { + t.Run(test.name, func(t *testing.T) { + rabbit := newTestRabbit(t, 2) + envelope, err := rabbit.deadLetterEnvelope([]byte(`{"traceId":"trace","payload":{"key":"clip.mp4"}}`), DeadLetterReasonHandlerError, 2) + if err != nil { + t.Fatal(err) + } + calls, attempts, reconnects := 0, 0, 0 + rabbit.options.DeadLetterObserver = func(event DeadLetterPublishEvent) { + calls++ + if !rabbit.publishMu.TryLock() { + t.Fatal("observer called under publishing lock") + } + rabbit.publishMu.Unlock() + if !rabbit.mu.TryLock() { + t.Fatal("observer called under state lock") + } + rabbit.mu.Unlock() + if attempts != test.wantAttempts { + t.Fatalf("observer before final attempt: %d", attempts) + } + want := DeadLetterPublishEvent{ + Source: "c", Destination: "c-dlq", Reason: DeadLetterReasonHandlerError, + Attempts: 2, TraceID: "trace", MediaFileName: "clip.mp4", + Confirmed: test.confirmed && test.wantErr == nil, Err: test.wantErr, + } + if event != want { + t.Fatalf("event = %+v, want %+v", event, want) + } + } + err = rabbit.publishAndObserveDeadLetter("c-dlq", envelope, test.confirmed, func() error { + return rabbitPublishWithReconnect(func() error { + return test.connectErr + }, func() error { + rabbit.publishMu.Lock() + defer rabbit.publishMu.Unlock() + if calls != 0 { + t.Fatal("observer called before publish finished") + } + attempts++ + if attempts == 1 { + return test.firstErr + } + return test.finalErr + }, func() bool { return false }, func() error { + reconnects++ + return test.reconnectErr + }) + }) + if !errors.Is(err, test.wantErr) || calls != 1 || attempts != test.wantAttempts { + t.Fatalf("err=%v calls=%d attempts=%d", err, calls, attempts) + } + if (reconnects == 1) != (test.firstErr == closed) { + t.Fatalf("reconnects=%d", reconnects) + } + }) + } +} + +func TestRabbitDeadLetterObserverCorrelation(t *testing.T) { + for _, test := range []struct { + name, body, trace, file string + }{ + {"pipeline key wins", `{"traceId":"trace","fileName":"other","payload":{"key":"clip.mp4","password":"SECRET"},"token":"SECRET"}`, "trace", "clip.mp4"}, + {"fileName fallback", `{"traceId":"trace","fileName":"clip.mp4"}`, "trace", "clip.mp4"}, + {"unrelated fields", `{"traceId":"trace","unknown":{"password":"SECRET"},"payload":{"key":"clip.mp4","nested":"SECRET"}}`, "trace", "clip.mp4"}, + {"raw malformed", `{"traceId":"trace","payload":{"key":"clip.mp4"}`, "", ""}, + {"invalid correlation type", `{"traceId":"trace","fileName":7}`, "", ""}, + {"raw binary", "\xff\x00", "", ""}, + {"null body", `null`, "", ""}, + {"null payload", `{"payload":null}`, "", ""}, + {"unrelated JSON", `{"password":"SECRET"}`, "", ""}, + } { + t.Run(test.name, func(t *testing.T) { + rabbit := newTestRabbit(t, 2) + // Historical envelopes can record a replay queue as Destination. + envelope, err := encodeDeadLetter([]byte(test.body), DeadLetterMetadata{ + Source: "original-source", Destination: "replay-queue", + Reason: DeadLetterReasonMalformed, Attempts: 7, + }) + if err != nil { + t.Fatal(err) + } + calls := 0 + rabbit.options.DeadLetterObserver = func(event DeadLetterPublishEvent) { + calls++ + if event.Source != "original-source" || event.Destination != "c-dlq" || + event.Reason != DeadLetterReasonMalformed || event.Attempts != 7 || + event.TraceID != test.trace || event.MediaFileName != test.file { + t.Fatalf("event = %+v", event) + } + if strings.Contains(fmt.Sprintf("%+v", event), "SECRET") { + t.Fatal("observer leaked unrelated payload fields") + } + } + if err := rabbit.publishAndObserveDeadLetter("c-dlq", envelope, true, func() error { return nil }); err != nil || calls != 1 { + t.Fatalf("err=%v calls=%d", err, calls) + } + }) + } +} + +func TestRabbitDeadLetterObserverEntryPoints(t *testing.T) { + for _, method := range []string{"AddToDeadletter", "Publish", "PublishDeadLetter"} { + t.Run(method, func(t *testing.T) { + rabbit := newTestRabbit(t, 2) + published, calls := false, 0 + rabbit.options = NewRabbitOptions(). + SetConsumerQueue("c"). + SetDeadletterQueue("c-dlq"). + SetDeadLetterObserver(func(event DeadLetterPublishEvent) { + calls++ + if !published || !event.Confirmed || event.Err != nil || event.TraceID != "trace" { + t.Fatalf("event = %+v, published=%v", event, published) + } + }).Build() + rabbit.deadLetterReplayPublish = func(context.Context, string, []byte) error { + published = true + return nil + } + payload := []byte(`{"traceId":"trace"}`) + var err error + switch method { + case "AddToDeadletter": + err = rabbit.AddToDeadletter(payload) + case "Publish": + err = rabbit.Publish("c-dlq", payload) + case "PublishDeadLetter": + err = rabbit.PublishDeadLetter(context.Background(), payload, DeadLetterMetadata{Source: "c", Destination: "c-dlq"}) + } + if err != nil || calls != 1 { + t.Fatalf("err=%v calls=%d", err, calls) + } + }) + } +} + +func TestRabbitDeadLetterObserverRuntimeFailures(t *testing.T) { + for _, method := range []string{"AddToDeadletter", "Publish", "PublishConfirmed", "PublishWithDelay", "forced transfer", "PublishDeadLetter"} { + t.Run(method, func(t *testing.T) { + rabbit := newTestRabbit(t, 2) + rabbit.options.ConfirmedDelivery = true + // Invalid URI fails before dialing, exercising the real public paths. + rabbit.connectionString = "://" + events := make(chan DeadLetterPublishEvent, 2) + rabbit.options.DeadLetterObserver = func(event DeadLetterPublishEvent) { events <- event } + payload := []byte("malformed") + var err error + switch method { + case "AddToDeadletter": + err = rabbit.AddToDeadletter(payload) + case "Publish": + err = rabbit.Publish("c-dlq", payload) + case "PublishConfirmed": + err = rabbit.PublishConfirmed("c-dlq", payload) + case "PublishWithDelay": + rabbit.PublishWithDelay("c-dlq", payload, 0) + case "forced transfer": + err = rabbit.addToDeadletterConfirmed(payload, DeadLetterReasonMalformed) + case "PublishDeadLetter": + err = rabbit.PublishDeadLetter(context.Background(), payload, DeadLetterMetadata{Source: "c", Destination: "c-dlq"}) + } + select { + case event := <-events: + if event.Err == nil || event.Confirmed || event.TraceID != "" || event.MediaFileName != "" { + t.Fatalf("event = %+v", event) + } + if method != "PublishWithDelay" && !errors.Is(err, event.Err) { + t.Fatalf("returned error=%v, event error=%v", err, event.Err) + } + case <-time.After(time.Second): + t.Fatal("missing publish outcome") + } + if len(events) != 0 { + t.Fatal("duplicate publish outcome") + } + }) + } +} + +func TestRabbitDeadLetterObserverRetryExhaustion(t *testing.T) { + for _, confirmed := range []bool{false, true} { + rabbit := newTestRabbit(t, 2) + calls := 0 + rabbit.options.DeadLetterObserver = func(event DeadLetterPublishEvent) { + calls++ + if event.Reason != DeadLetterReasonRetryExhausted || event.Attempts != 2 || event.Confirmed != confirmed { + t.Fatalf("event = %+v", event) + } + } + publish := func(destination string, body []byte, _ amqp.Table) error { + return rabbit.publishAndObserveDeadLetter(destination, body, confirmed, func() error { return nil }) + } + if err := rabbit.retryOrDeadletter(nil, []byte("raw"), time.Nanosecond, publish); err != nil || calls != 0 { + t.Fatalf("ordinary retry err=%v calls=%d", err, calls) + } + if err := rabbit.retryOrDeadletter(amqp.Table{retryCountHeader: int32(2)}, []byte("raw"), time.Nanosecond, publish); err != nil || calls != 1 { + t.Fatalf("exhausted retry err=%v calls=%d", err, calls) + } + } +} + +func TestRabbitDeadLetterObserverSettlement(t *testing.T) { + for _, reason := range []DeadLetterReason{DeadLetterReasonMalformed, DeadLetterReasonHandlerError, DeadLetterReasonPublishFailed} { + t.Run(string(reason), func(t *testing.T) { + rabbit := newTestRabbit(t, 2) + ack := &recordingAcknowledger{} + calls := 0 + rabbit.options.DeadLetterObserver = func(event DeadLetterPublishEvent) { + calls++ + if event.Reason != reason || ack.acks != 0 { + t.Fatalf("event=%+v acks=%d", event, ack.acks) + } + } + rabbit.deadLetterReplayPublish = func(context.Context, string, []byte) error { return nil } + if err := rabbit.deadletterAndSettle(amqp.Delivery{Acknowledger: ack}, []byte("raw"), reason, false); err != nil || calls != 1 || ack.acks != 1 { + t.Fatalf("err=%v calls=%d acks=%d", err, calls, ack.acks) + } + }) + } +} + +func TestRabbitDeadLetterObserverFailureBeforeRecovery(t *testing.T) { + rabbit := newTestRabbit(t, 2) + rabbit.options.ConfirmedDelivery = true + rabbit.connectionString = "://" + ack := &recordingAcknowledger{} + calls, recovered := 0, false + rabbit.options.DeadLetterObserver = func(event DeadLetterPublishEvent) { + calls++ + if event.Err == nil || event.Confirmed || recovered || ack.acks != 0 { + t.Fatalf("event=%+v recovered=%v acks=%d", event, recovered, ack.acks) + } + } + rabbit.SetDisasterRecoveryHandler(func([]byte) error { + if calls != 1 { + t.Fatal("missing publish failure before recovery") + } + recovered = true + return nil + }) + err := rabbit.deadletterAndSettle(amqp.Delivery{Acknowledger: ack}, []byte("raw"), DeadLetterReasonHandlerError, true) + if err != nil || !recovered || calls != 1 || ack.acks != 1 || ack.nacks != 0 { + t.Fatalf("err=%v recovered=%v calls=%d ack=%+v", err, recovered, calls, ack) + } +} + +func TestRabbitDeadLetterObserverIgnoresAdministrativeRestoration(t *testing.T) { + for _, operation := range []string{"inspect", "scan", "skip", "replay", "discard", "transform failure", "restore failure"} { + t.Run(operation, func(t *testing.T) { + rabbit := newTestRabbit(t, 2) + rabbit.options.DeadLetterObserver = func(DeadLetterPublishEvent) { t.Fatal("administration emitted a new entry") } + body, err := rabbit.deadLetterEnvelope([]byte("raw"), DeadLetterReasonHandlerError, 0) + if err != nil { + t.Fatal(err) + } + ack := &fakeRabbitAcknowledger{} + read := false + rabbit.deadLetterGet = func() (amqp.Delivery, bool, error) { + if read { + return amqp.Delivery{}, false, nil + } + read = true + return amqp.Delivery{Acknowledger: ack, DeliveryTag: 1, Body: body}, true, nil + } + publishes := 0 + rabbit.deadLetterReplayPublish = func(context.Context, string, []byte) error { + publishes++ + if operation == "restore failure" { + return errors.New("restore failed") + } + return nil + } + if operation == "inspect" || operation == "restore failure" { + _, err = rabbit.InspectDeadLetters(context.Background(), DeadLetterInspectRequest{Limit: 1}) + } else { + request := DeadLetterReplayRequest{Limit: 1, Execute: operation != "scan"} + switch operation { + case "skip", "discard", "transform failure": + request.Transform = func(context.Context, []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { + if operation == "transform failure" { + return nil, errors.New("transform failed") + } + return []DeadLetterReplayTransformation{{Skip: operation == "skip", Discard: operation == "discard"}}, nil + } + } + _, err = rabbit.ReplayDeadLetters(context.Background(), request) + } + wantError := operation == "transform failure" || operation == "restore failure" + if (err != nil) != wantError { + t.Fatalf("err=%v", err) + } + wantPublishes := 1 + if operation == "discard" { + wantPublishes = 0 + } + if publishes != wantPublishes { + t.Fatalf("publishes=%d, want %d", publishes, wantPublishes) + } + }) + } +} + +func TestRabbitDeadLetterObserverDisabled(t *testing.T) { + rabbit := newTestRabbit(t, 2) + if rabbit.options.DeadLetterObserver != nil { + t.Fatal("observer enabled by default") + } + failed := errors.New("publish failed") + if err := rabbit.publishAndObserveDeadLetter("c-dlq", []byte("invalid envelope"), true, func() error { return failed }); err != failed { + t.Fatalf("err=%v", err) + } +} diff --git a/pkg/queue/deadletter_rabbitmq.go b/pkg/queue/deadletter_rabbitmq.go index 042c97d..2c4f5fb 100644 --- a/pkg/queue/deadletter_rabbitmq.go +++ b/pkg/queue/deadletter_rabbitmq.go @@ -17,12 +17,14 @@ func (r *RabbitMQ) PublishDeadLetter(ctx context.Context, payload []byte, metada if err != nil { return err } - if r.deadLetterReplayPublish == nil { - if err := r.ensureConnected(); err != nil { - return err + return r.publishAndObserveDeadLetter(r.options.DeadletterQueue, envelope, true, func() error { + if r.deadLetterReplayPublish == nil { + if err := r.ensureConnected(); err != nil { + return err + } } - } - return r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, envelope) + return r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, envelope) + }) } // ReadMessagesToDeadletter transfers raw consumer deliveries into versioned diff --git a/pkg/queue/rabbitmq.go b/pkg/queue/rabbitmq.go index 5d2fbda..8e420fa 100644 --- a/pkg/queue/rabbitmq.go +++ b/pkg/queue/rabbitmq.go @@ -51,6 +51,10 @@ type RabbitOptions struct { // consumer queue forever (see retryOrDeadletter). MaxRetries int + // DeadLetterObserver receives final DLQ publish outcomes, outside publishing + // locks. It must be concurrency-safe, return promptly, and not panic. + DeadLetterObserver func(DeadLetterPublishEvent) + // TLS configuration for secure connections (e.g., AWS Amazon MQ) TLS bool // Enable TLS (auto-enabled when host starts with amqps://) TLSInsecureSkipVerify bool // Skip TLS certificate verification (development only) @@ -105,6 +109,12 @@ func (b *RabbitOptionsBuilder) SetDeadletterQueue(queueName string) *RabbitOptio return b } +// SetDeadLetterObserver sets the optional observer for final DLQ publish outcomes. +func (b *RabbitOptionsBuilder) SetDeadLetterObserver(observer func(DeadLetterPublishEvent)) *RabbitOptionsBuilder { + b.options.DeadLetterObserver = observer + return b +} + // SetRouterQueue sets the router queue name func (b *RabbitOptionsBuilder) SetRouterQueue(queueName string) *RabbitOptionsBuilder { b.options.RouterQueue = queueName @@ -1147,17 +1157,11 @@ func (r *RabbitMQ) PublishConfirmed(queueName string, payload []byte) error { } func (r *RabbitMQ) publishLegacyWithReconnect(queueName string, payload []byte, headers amqp.Table) error { - if err := r.ensureConnected(); err != nil { - return err - } - - err := r.publishLegacy(queueName, payload, headers) - if err != nil && (r.needsReconnect() || isClosedError(err)) { - if reconnectErr := r.Reconnect(); reconnectErr == nil { - err = r.publishLegacy(queueName, payload, headers) - } - } - return err + return r.publishAndObserveDeadLetter(queueName, payload, false, func() error { + return rabbitPublishWithReconnect(r.ensureConnected, func() error { + return r.publishLegacy(queueName, payload, headers) + }, r.needsReconnect, r.Reconnect) + }) } func (r *RabbitMQ) publishLegacyWithRecovery(queueName string, payload []byte, headers amqp.Table) error { @@ -1188,14 +1192,22 @@ func (r *RabbitMQ) publishLegacy(queueName string, payload []byte, headers amqp. // closed/stale channel, reconnects once and retries. It is the shared body // behind confirmed publishing and confirmed consumer transfers. func (r *RabbitMQ) publishConfirmedWithReconnect(queueName string, payload []byte, headers amqp.Table) error { - if err := r.ensureConnected(); err != nil { + return r.publishAndObserveDeadLetter(queueName, payload, true, func() error { + return rabbitPublishWithReconnect(r.ensureConnected, func() error { + return r.publishConfirmed(queueName, payload, headers) + }, r.needsReconnect, r.Reconnect) + }) +} + +func rabbitPublishWithReconnect(ensureConnected, publish func() error, needsReconnect func() bool, reconnect func() error) error { + if err := ensureConnected(); err != nil { return err } - err := r.publishConfirmed(queueName, payload, headers) - if err != nil && (r.needsReconnect() || isClosedError(err)) { - if reconnectErr := r.Reconnect(); reconnectErr == nil { - err = r.publishConfirmed(queueName, payload, headers) + err := publish() + if err != nil && (needsReconnect() || isClosedError(err)) { + if reconnectErr := reconnect(); reconnectErr == nil { + err = publish() } } return err @@ -1355,7 +1367,9 @@ func (r *RabbitMQ) addToDeadletter(payload []byte, reason DeadLetterReason, atte if r.deadLetterReplayPublish != nil { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - return r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, envelope) + return r.publishAndObserveDeadLetter(r.options.DeadletterQueue, envelope, true, func() error { + return r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, envelope) + }) } if r.options.ConfirmedDelivery { return r.publishConfirmedWithReconnect(r.options.DeadletterQueue, envelope, nil)