Sending
What a publish guarantees, what it does not, and the topology that decides whether anybody ever reads the message.
Publish, and what err == nil buys
The publisher runs in confirm mode, so a nil error means the broker has taken responsibility for the message — not merely that it left the process. It still does not mean any queue received it, which is what Mandatory is for.
go
package main
import (
"errors"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/errmsgs"
)
// --- Example 1: publishing ---------------------------------------------------
//
// The publisher runs in confirm mode, so one publish is one round trip that ends
// in the broker's acknowledgement. That is what makes `err == nil` worth
// something here: the broker has taken responsibility for the message, not
// merely accepted the bytes. It is the difference between *sent* and *stored*,
// and the reason persistence alone never guaranteed anything.
//
// What nil still does not say is that anything will ever *read* it. A message
// published to an exchange with no matching binding is confirmed and thrown
// away. Routing is topology's job (02), not the publisher's — Mandatory is the
// opt-in that turns that particular silence into an error.
// Order is the payload. Anything that is not []byte is JSON-encoded on the way
// out, with content type application/json.
type Order struct {
ID string `json:"id"`
Total int64 `json:"total"`
Currency string `json:"currency"`
PlacedAt time.Time `json:"placed_at"`
}
const ordersExchange = "orders"
// publishOrderCreated is the whole API for the common case.
//
// Note what has to have happened before it: the row exists. A message published
// inside a transaction escapes immediately — the broker knows nothing about the
// database's transaction — so a consumer can win the race and go looking for an
// order that is not there yet. Commit first, publish second.
func publishOrderCreated(ctx core.IContext, order Order) core.IError {
// PublishAs is Publish with the payload type pinned by the compiler. Same
// call, same wire format; worth having on a publish nobody reads again for a
// year and then changes the struct of.
return core.PublishAs(ctx.MQ(), ordersExchange, "order.created", order)
}
// publishOrderPaid spells out the properties that decide whether this message is
// debuggable and de-duplicatable once it is somebody else's problem.
func publishOrderPaid(ctx core.IContext, order Order, requestID string) core.IError {
return ctx.MQ().PublishWith(ordersExchange, "order.paid", order, core.PublishOptions{
// MessageID has to come from the identity of the *work* — an order id, a
// payment id — never a fresh UUID per attempt. Two publishes of the same
// event must collide, or the consumer's dedupe has nothing to match on.
MessageID: order.ID,
// CorrelationID threads request → message → consumer work together in the
// logs. It costs nothing and is the first thing wanted during an incident.
CorrelationID: requestID,
// Type names the event for a queue that receives several kinds.
Type: "order.paid",
// Headers travel with the message. A schema version from day one is where
// a breaking change gets to stand later.
Headers: map[string]any{"schema": 1},
// Mandatory costs an extra round trip and buys MQ_NO_ROUTE instead of a
// silent discard. Right for an event whose loss becomes a support ticket;
// wrong for a fan-out that is *meant* to have no listeners yet, where the
// error is one nobody can act on.
Mandatory: true,
})
}
// handlePublishError tells apart the failures that mean different things. The
// one that always deserves a decision is MQ_NACK: the broker accepted the
// message and then refused it, so it was *not* stored and the caller is the only
// one who can say what happens next.
func handlePublishError(ctx core.IContext, err core.IError) error {
switch {
case err == nil:
return nil
case errors.Is(err, core.ErrMQNoRoute):
// Confirmed and discarded: nothing is bound for this key. Almost always a
// topology that was never applied, not a transient fault — retrying sends
// it to the same nowhere.
return ctx.NewError(err, errmsgs.MQError)
case errors.Is(err, core.ErrMQNack):
// The broker took it and then said no (a full disk, a queue refusing the
// write). The message is gone; hand it to whatever can send it again —
// an outbox row (05), a job — rather than swallowing the error.
return ctx.NewError(err, errmsgs.MQError)
case errors.Is(err, core.ErrMQDisabled):
// No MQ_* configuration at all. The queue fails loudly where the cache
// would degrade quietly, because a dropped message is work somebody
// believes was handed off and nobody will ever pick up.
return ctx.NewError(err, errmsgs.MQError)
case errors.Is(err, core.ErrMQClosed):
// Shutting down. A publish that arrives now is better answered honestly
// than allowed to hold the shutdown open.
ctx.Log().Warn("mq: publish during shutdown", "err", err)
return err
default:
// Dial failures, timeouts, waiting too long for a free channel. Transient
// by nature — this is the group worth retrying.
return err
}
}Exchanges, queues and bindings at boot
Declaring is idempotent, so deploying the service is what creates its topology. Includes the two things most often left out: a dead-letter exchange with a queue actually bound to it, and queue depth exported as the metric that moves first.
go
package main
import (
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 2: topology, declared at boot -----------------------------------
//
// Declaring is idempotent, so this runs on every boot: deploying the service is
// what creates its topology, and nothing depends on somebody having run a script
// or clicked in the management UI first. Declaring something that already exists
// with *different* settings is an error from the broker, which is the feature:
// it catches topology that changed in code while production still has the old
// shape, instead of letting two environments drift apart quietly.
//
// Who declares what:
//
// publisher only the exchanges it sends into
// consumer its own queue, its own bindings, its own DLX (see 03)
//
// A queue belongs to whoever reads it. A publisher that declares somebody else's
// queue is a service that must be redeployed whenever that consumer changes a
// binding, and the reason an obsolete queue still fills up months later.
const (
ordersDLX = "orders.dlx"
ordersDeadQ = "orders.dead"
shippingQueue = "orders.shipping"
paymentsQueue = "orders.payments"
)
// declarePublisherTopology is everything the publishing side needs to know.
func declarePublisherTopology(app *core.App) core.IError {
return app.MQ().DeclareExchange(core.ExchangeConfig{
Name: ordersExchange,
// topic subsumes direct (an exact key) and fanout ("#"), so it is the
// default worth choosing: a second consumer can be added later with a new
// binding and no change here at all.
Kind: core.ExchangeTopic,
// Transient is false, so this is durable. The zero value is the safe one
// on purpose — an exchange that vanishes with the broker is almost never
// what anybody meant.
})
}
// declareDeadLetterTopology declares the DLX *and* the queue that gives it a
// point.
//
// A dead-letter exchange with nothing bound to it is a bin: the broker routes
// the rejected message into it and drops it, exactly as if there were no DLX.
// The cost of having one is an empty queue; the cost of not having one is the
// unanswerable question "where did that order go".
func declareDeadLetterTopology(app *core.App) core.IError {
mq := app.MQ()
if err := mq.DeclareExchange(core.ExchangeConfig{
Name: ordersDLX,
Kind: core.ExchangeTopic,
}); err != nil {
return err
}
if _, err := mq.DeclareQueue(core.QueueConfig{
Name: ordersDeadQ,
// Two weeks to come and look, rather than for ever. A dead-letter queue
// nobody ever empties is a disk that fills at the worst possible moment.
TTL: 14 * 24 * time.Hour,
}); err != nil {
return err
}
// "#" catches every routing key that dead-letters into this exchange.
return mq.BindQueue(ordersDeadQ, ordersDLX, "#")
}
// declarePaymentsQueue is the queue where losing one message means somebody
// reconciling by hand.
func declarePaymentsQueue(app *core.App) core.IError {
mq := app.MQ()
info, err := mq.DeclareQueue(core.QueueConfig{
Name: paymentsQueue,
DeadLetterExchange: ordersDLX,
// Replicated across the cluster, so losing a node loses no message. It
// costs throughput and memory, which is why it belongs on the queues that
// carry money and not on the one that sends notifications.
//
// Known limits: a quorum queue supports neither MaxPriority nor Exclusive.
Quorum: true,
})
if err != nil {
return err
}
// DeclareQueue reports the queue as it is at that moment — a free reading of
// what survived the last deploy.
app.Log().Info("payments queue declared",
"queue", info.Name, "messages", info.Messages, "consumers", info.Consumers)
return mq.BindQueue(paymentsQueue, ordersExchange, "order.paid")
}
// queueDepth reads depth without declaring anything: QueueInfo uses a passive
// declare, so it creates nothing and fails when the queue is absent. A typo
// therefore reports an error instead of conjuring a ghost queue that quietly
// collects messages nobody consumes.
//
// Depth is the signal that arrives first when a consumer falls behind. It is
// still succeeding, only too slowly, so there is no error rate to alert on —
// export this, not just failures.
func queueDepth(app *core.App, queues ...string) {
for _, q := range queues {
info, err := app.MQ().QueueInfo(q)
if err != nil {
app.Log().Warn("mq: cannot read queue depth", "queue", q, "err", err)
continue
}
app.Log().Info("mq queue depth",
"queue", info.Name, "messages", info.Messages, "consumers", info.Consumers)
}
}