Receiving
The handler, the acknowledgement it implies, and what to do with a failure that will not fix itself in the next millisecond.
Consuming and acknowledging
Return nil to ack, an error to dead-letter, core.Requeue to ask for it again. The default not being requeue is the whole design: a permanent failure that comes straight back loops as fast as the broker can deliver it.
go
package main
import (
"context"
"database/sql"
"errors"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 3: consuming ----------------------------------------------------
//
// A consumer turns queues into handlers the way the HTTP server turns routes
// into handlers, and owns everything nobody wants to write twice: the
// connection, the prefetch, the acknowledgement, a context per message, panic
// recovery, reconnection after a broker restart, and the drain on shutdown.
//
// Start connects before it returns, so a broker that is down is an error at boot
// that the caller can act on — not a goroutine retrying into a log nobody reads
// during a deploy.
func newShippingConsumer(app *core.App) core.IMQConsumer {
c := app.NewMQConsumer(
// Prefetch is the backpressure knob: how many unacknowledged messages the
// broker will push at this instance. Too low and the consumer waits on the
// network between messages; too high and one instance hoards work another
// replica is sitting idle for. prefetch ≈ concurrency × 2 is a starting
// point that holds up.
core.WithMQPrefetch(8),
// 1 by default, which is what keeps a queue's messages in order. Raising
// it trades that order for throughput and nothing else — do it only when
// two messages about the same entity can be handled either way round and
// still leave the right state behind.
core.WithMQConcurrency(4),
// The ceiling on one handler. Make it longer than the slowest thing the
// handler genuinely does, or every slow message dead-letters on a timeout
// that was never realistic.
core.WithMQHandlerTimeout(30*time.Second),
// What the management UI shows next to this connection.
core.WithMQConsumerTag("examples-shipping"),
)
// OnQueue declares the exchange, the queue and the bindings before consuming
// — and again after every reconnect. On(queue, handler) is the other form,
// for a queue somebody else owns; it logs as topology=external precisely
// because nothing here will recreate it.
c.OnQueue(core.ConsumeQueue{
Queue: core.QueueConfig{
Name: shippingQueue,
DeadLetterExchange: ordersDLX,
},
Exchange: &core.ExchangeConfig{Name: ordersExchange, Kind: core.ExchangeTopic},
BindingKeys: []string{"order.created", "order.paid"},
}, handleShipping)
return c
}
// handleShipping is the acknowledgement contract in one function:
//
// return nil ack — the broker forgets the message
// return err reject, no requeue → the DLX (or gone, without one)
// return core.Requeue(err) reject and redeliver, immediately
// panic recovered, then treated exactly like a returned error
//
// The default not being requeue is deliberate. A permanent failure that is
// requeued comes straight back, fails the same way, and loops as fast as the
// broker can deliver — the classic way a consumer takes its database down with
// it.
func handleShipping(ctx core.IMQContext, d *core.Delivery) error {
order, err := core.BindDelivery[Order](d)
if err != nil {
// A body that will not parse will not parse better the second time. This
// is the archetypal dead-letter: keep the evidence, do not retry it.
return err
}
// At-least-once is the only guarantee on offer, so a handler has to survive
// running twice. Redelivered says this *may* be a repeat — it is a hint and
// never a proof, because a republished copy (04) arrives with it false.
ctx.Log().Info("shipping order",
"order", order.ID, "key", d.RoutingKey, "redelivered", d.Redelivered)
if err := ship(ctx, order); err != nil {
if errors.Is(err, sql.ErrConnDone) || errors.Is(err, context.DeadlineExceeded) {
// A failure about *now*: the same message succeeds once the database
// is back. Requeue has no counter and no delay though, so this is only
// right for a stumble measured in milliseconds — 04 is the version
// with both.
return core.Requeue(err)
}
return err
}
return nil
}
// ship stands in for the real work. Making it idempotent — a unique index plus
// an upsert on the order id — is what turns a redelivery into a no-op.
func ship(ctx core.IMQContext, order Order) error {
if order.ID == "" {
// A business rule saying no is not an error: it must not go and sit in the
// dead-letter queue where somebody will investigate it.
ctx.Log().Warn("mq: order without an id, dropping")
return nil
}
return nil
}
// stopShippingConsumer stops one consumer early — draining work on a rolling
// deploy while the process keeps serving HTTP.
//
// It is not needed at shutdown: the App remembers every consumer it handed out
// and stops them *before* closing the pools their handlers are using.
func stopShippingConsumer(app *core.App, c core.IMQConsumer) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
// The order inside Stop: cancel at the broker so no new deliveries arrive →
// nack-with-requeue anything already taken but not started → wait for
// in-flight handlers → at the deadline, close underneath them so their
// messages return to the broker unacknowledged rather than hold shutdown
// open. Keep that deadline below the orchestrator's grace period, or none of
// it ever happens.
if err := c.Stop(ctx); err != nil {
app.Log().Error("mq consumer did not drain in time", "err", err)
}
}Retry with a real gap, and a replayable DLQ
A delay queue — TTL plus a dead-letter exchange pointing home — gives retries an actual interval, and an attempt counter in a header of our own gives them a ceiling. What lands in the DLQ is then only what genuinely ran out of tries.
go
package main
import (
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 4: retry with a real gap, and a DLQ you can replay --------------
//
// core.Requeue means "again, now". No counter, no delay. That is right for a
// stumble measured in milliseconds and wrong for a dependency that is down for
// ten minutes: the message comes straight back, fails the same way, and the loop
// runs as fast as the broker can deliver it.
//
// The pattern that works is a delay queue — a queue nobody consumes, with a TTL
// and a dead-letter exchange pointing back at the original one. A message sent
// there lies still for the TTL and then bounces home by itself, with its
// original routing key, which is what makes the round trip invisible to the
// handler.
//
// orders ──▶ orders.shipping ──(transient)──▶ orders.retry.10s
// ▲ │ TTL 10s
// └──────────── DLX back into orders ◀───────────┘
const (
retryExchange = "orders.retry"
maxAttempts = 5
)
type retryLevel struct {
queue string
delay time.Duration
}
// retryLevels is one queue per delay, deliberately.
//
// The alternative — one queue and a different PublishOptions.Expiration per
// message — looks simpler and is a trap: RabbitMQ only expires a message when it
// reaches the head of the queue, so a 10s message queued behind a 10m one waits
// ten minutes (head-of-line blocking).
func retryLevels() []retryLevel {
return []retryLevel{
{queue: "orders.retry.10s", delay: 10 * time.Second},
{queue: "orders.retry.1m", delay: time.Minute},
{queue: "orders.retry.10m", delay: 10 * time.Minute},
}
}
func declareRetryTopology(app *core.App) core.IError {
mq := app.MQ()
if err := mq.DeclareExchange(core.ExchangeConfig{
Name: retryExchange,
Kind: core.ExchangeTopic,
}); err != nil {
return err
}
for _, lvl := range retryLevels() {
if _, err := mq.DeclareQueue(core.QueueConfig{
Name: lvl.queue,
TTL: lvl.delay,
// Where an expired message goes: back into the exchange the work came
// from, so it is redelivered to the normal handler.
DeadLetterExchange: ordersExchange,
}); err != nil {
return err
}
if err := mq.BindQueue(lvl.queue, retryExchange, lvl.queue); err != nil {
return err
}
}
return nil
}
// scheduleRetry acks the delivery it was given (by returning nil) and puts a
// copy in the delay queue.
//
// Publish first, ack second: this order can duplicate the message if the process
// dies in between, and the other order loses it. Duplication is the failure a
// consumer already has to tolerate; loss is not.
func scheduleRetry(ctx core.IMQContext, d *core.Delivery, cause error) error {
attempt := attemptOf(d) + 1
if attempt > maxAttempts {
// Out of attempts: return the cause so the message rejects into the DLX.
// What waits there is then only what genuinely exhausted its retries,
// which is what makes the dead-letter queue worth reading.
return cause
}
levels := retryLevels()
lvl := levels[min(attempt-1, len(levels)-1)]
ctx.Log().Warn("mq: retrying later",
"attempt", attempt, "in", lvl.queue, "delay", lvl.delay.String(), "err", cause)
return ctx.MQ().PublishWith(retryExchange, lvl.queue, d.Body, core.PublishOptions{
// d.Body is []byte, and this package will not claim on the caller's behalf
// that a byte slice is JSON. Without this the copy arrives as
// application/octet-stream and the next handler is none the wiser.
ContentType: d.ContentType,
// Keep the identity, or the consumer's dedupe stops recognising its own
// message.
MessageID: d.MessageID,
CorrelationID: d.CorrelationID,
Type: d.Type,
Headers: withAttempt(d.Headers, attempt),
})
}
// attemptOf counts with a header of our own. The broker's x-death array carries
// something similar, but reading it means reaching into the amqp091 types this
// layer exists to keep out of handlers — and it counts "times dead-lettered from
// this queue", which is not the same question.
//
// AMQP header integers arrive as whichever width the wire used, so all three
// cases are real.
func attemptOf(d *core.Delivery) int {
switch n := d.Headers["x-attempt"].(type) {
case int32:
return int(n)
case int64:
return int(n)
case int:
return n
default:
return 0
}
}
// withAttempt copies the headers rather than mutating the delivery's map, so the
// original message stays exactly as it arrived for logging and Sentry.
func withAttempt(h map[string]any, n int) map[string]any {
out := make(map[string]any, len(h)+1)
for k, v := range h {
out[k] = v
}
out["x-attempt"] = n
return out
}
// newDLQReplayConsumer reads the dead-letter queue and republishes into the
// original exchange. Start it only once the cause is fixed — replaying a full
// DLQ into a still-broken consumer refills it, with a second helping of load.
func newDLQReplayConsumer(app *core.App) core.IMQConsumer {
replay := app.NewMQConsumer(
core.WithMQConcurrency(1),
core.WithMQConsumerTag("dlq-replay"),
)
replay.On(ordersDeadQ, func(ctx core.IMQContext, d *core.Delivery) error {
// Not d.Exchange: a dead-lettered delivery reports the exchange it was
// last published to, which is the DLX itself. Republishing there would
// route straight back into this queue and spin. The destination has to be
// named.
return ctx.MQ().PublishWith(ordersExchange, d.RoutingKey, d.Body, core.PublishOptions{
ContentType: d.ContentType,
MessageID: d.MessageID,
CorrelationID: d.CorrelationID,
Type: d.Type,
// Start the count again — this message is getting a fresh budget.
Headers: withAttempt(d.Headers, 0),
})
})
return replay
}