Message Queue (RabbitMQ)

core.IMQ — RabbitMQ publisher (amqp091). ctx.MQ() returns a handle bound to the request context.

Connect & wire

mq, err := core.NewMQ(env)   // MQ_* / MQ_CONNECTION_STRING
app, _ := core.NewApp(env, core.WithMQ(mq))

MQ_CONNECTION_STRING=amqp://user:pass@host:5672/vhost   # amqps:// for TLS
# or
MQ_HOST=... MQ_PORT=... MQ_USER=... MQ_PASSWORD=...

Publishing

// payload is JSON-encoded unless it is []byte
err := ctx.MQ().Publish("orders", "order.created", Order{ID: "o1"})

// typed helper
err = core.PublishAs(ctx.MQ(), "orders", "order.created", order)

Messages are published as persistent, application/json, with a timestamp.

Interface

type IMQ interface {
    Publish(exchange, key string, msg any) IError
    WithContext(ctx context.Context) IMQ
    Close() IError
}

Notes

• The connection/channel live on the App; app.Shutdown() closes them.
• Each publish is bounded by a short timeout derived from the request context.
• Consuming is not part of this publisher; run consumers with the amqp091 client directly (or a dedicated worker) and build an IContext per delivery via app.NewContext(deliveryCtx, core.ModeMQ).