README
¶
babelqueue-go/machinery
machinery adapter for the BabelQueue Go runtime — "Polyglot Queues, Simplified."
machinery is a popular Redis/AMQP-backed Go task queue. This adapter makes a
machinery app produce and consume the canonical BabelQueue envelope, so it
interoperates byte-for-byte with the PHP/Laravel, Python, … SDKs. machinery is not
a new broker binding (its storage is Redis/AMQP, §1/§2) —
it is a framework adapter, the sibling of the …/asynq module.
machinery is argument-based (a task is a tasks.Signature with a Name and
typed Args), so the mapping is:
| BabelQueue | machinery |
|---|---|
job (the URN) |
task Name (machinery routes on it, like every BabelQueue consumer) |
| canonical envelope JSON | a single string Arg (byte-identical — no wrapping, no added fields) |
Install
go get github.com/babelqueue/babelqueue-go/machinery
Requires Go 1.23+.
Produce
import (
babelqueue "github.com/babelqueue/babelqueue-go"
bqmachinery "github.com/babelqueue/babelqueue-go/machinery"
)
// server is your configured *machinery/v2.Server
p := bqmachinery.NewProducer(server)
p.Dispatch(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042},
babelqueue.WithQueue("orders"), // recorded in meta.queue
)
meta.queue records the logical BabelQueue queue; machinery's own broker queue
/ routing key is set on the server (or on the signature) and need not equal it. For
machinery-native delivery options (RetryCount, ETA for delay, RoutingKey),
build the signature with Signature/SignatureFor, set the fields, then send it
with server.SendTaskWithContext.
Consume — route by URN
bqmachinery.Register(server, "urn:babel:orders:created",
func(ctx context.Context, env babelqueue.Envelope) error {
// env.Data, env.TraceID … (trace_id preserved across the hop)
return nil
})
Register registers the handler under the URN as the machinery task Name, so
machinery routes tasks whose name equals the URN to it. The bound task function
decodes the envelope argument, rejects one that fails consumer-side validation
(ErrNotAccepted, so machinery applies its own retry policy), and invokes the
handler — the machinery-native equivalent of BabelQueue's URN routing. For finer
control, Bind(handler) returns the plain func(ctx, string) error machinery task
func, and Signature/Envelope convert between an envelope and a
*tasks.Signature directly.
How it maps to the contract
The envelope is the byte-identical envelope argument (Redis/AMQP
§1/§2 payload identity); the URN is the machinery task
Name. The envelope is unchanged (schema_version stays 1); machinery is purely
additive.
Test
go test ./... # unit + conformance (fake Sender/Registrar seams — no broker)
The machinery server is reached through small Sender/Registrar seams, so the
suite runs network-free.
License
MIT
Documentation
¶
Overview ¶
Package machinery bridges the github.com/RichardKnop/machinery task queue to the BabelQueue canonical envelope, so a machinery-based Go app interoperates byte-for-byte with the PHP/Laravel, Python, ... SDKs.
machinery is not a BabelQueue broker binding (its Redis/AMQP storage is §1/§2); it is a framework adapter, mirroring the sibling github.com/babelqueue/babelqueue-go/asynq module. machinery is **argument-based** (a task is a tasks.Signature with a Name and typed Args), unlike asynq's raw-payload model, so the mapping is:
- the machinery task Name IS the envelope URN (job), so machinery routes on the URN the same way every BabelQueue consumer does, and
- a single string Arg carries the canonical envelope JSON byte-for-byte (no wrapping, no added fields — Redis §1 payload identity).
Produce wraps babelqueue.Make/Encode into a *tasks.Signature and sends it; consume decodes the signature's envelope arg back into a babelqueue.Envelope and dispatches it through a handler registered under the URN.
// produce
p := machinery.NewProducer(server) // server is *RichardKnop/machinery/v2.Server
p.Dispatch(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042},
babelqueue.WithQueue("orders"))
// consume — register a handler under the URN as the machinery task name
machinery.Register(server, "urn:babel:orders:created",
func(ctx context.Context, env babelqueue.Envelope) error {
// env.Data, env.TraceID ...
return nil
})
The envelope is unchanged (schema_version stays 1); machinery is purely additive.
Full spec: https://babelqueue.com
Index ¶
- Variables
- func Bind(handler babelqueue.Handler) func(ctx context.Context, body string) error
- func Envelope(sig *tasks.Signature) (babelqueue.Envelope, error)
- func Register(server Registrar, urn string, handler babelqueue.Handler) error
- func Signature(env babelqueue.Envelope) (*tasks.Signature, error)
- func SignatureFor(urn string, data map[string]any, opts ...babelqueue.Option) (*tasks.Signature, error)
- type Producer
- type Registrar
- type Sender
Constants ¶
This section is empty.
Variables ¶
var ErrNoEnvelopeArg = errors.New("babelqueue/machinery: signature has no envelope argument")
ErrNoEnvelopeArg is returned by Envelope when a signature carries no string envelope argument (so it was not produced by this adapter).
var ErrNotAccepted = errors.New("babelqueue/machinery: task payload is not an acceptable envelope")
ErrNotAccepted is returned by Bind/Register handlers when an inbound task payload fails consumer-side validation (babelqueue.Envelope.Accepts) — a missing URN, an unsupported schema_version, a blank trace_id, or missing data.
Functions ¶
func Bind ¶
Bind adapts a BabelQueue babelqueue.Handler to a machinery task function: machinery invokes it with the injected context and the envelope-JSON argument; it decodes the envelope, rejects it with ErrNotAccepted when it fails babelqueue.Envelope.Accepts (so machinery applies its own retry policy), and otherwise invokes handler. Use Register to wire it under the URN.
func Envelope ¶
func Envelope(sig *tasks.Signature) (babelqueue.Envelope, error)
Envelope decodes a machinery signature's envelope argument into a BabelQueue envelope, resolving the urn alias (see babelqueue.Decode). The signature Name is the URN, but the authoritative identity is the decoded body.
func Register ¶
func Register(server Registrar, urn string, handler babelqueue.Handler) error
Register wires handler onto the server for the given URN: it registers Bind(handler) under the URN as the machinery task name, so machinery routes tasks whose Name equals the URN to it — the machinery-native equivalent of BabelQueue's URN routing.
func Signature ¶
func Signature(env babelqueue.Envelope) (*tasks.Signature, error)
Signature builds a machinery task signature from an already-built envelope: Name = the URN (env.Job) and a single string Arg = the canonical envelope JSON. Set advanced machinery fields (RetryCount, ETA for delayed delivery, RoutingKey) on the returned signature before sending. It returns babelqueue.ErrEmptyURN when the envelope has no URN, and surfaces an encode error otherwise.
func SignatureFor ¶
func SignatureFor(urn string, data map[string]any, opts ...babelqueue.Option) (*tasks.Signature, error)
SignatureFor builds the canonical envelope for (urn, data) via babelqueue.Make and wraps it as a machinery signature (see Signature). It mints a fresh trace id unless babelqueue.WithTraceID is given.
Types ¶
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer publishes BabelQueue envelopes as machinery tasks over a Sender.
func NewProducer ¶
NewProducer wraps a Sender (a *machinery/v2.Server, or a fake in tests).
func (*Producer) Dispatch ¶
func (p *Producer) Dispatch(ctx context.Context, urn string, data map[string]any, opts ...babelqueue.Option) (*result.AsyncResult, error)
Dispatch builds the canonical envelope for (urn, data) and sends it as a machinery task routed by URN, returning machinery's AsyncResult. Pass envelope options (babelqueue.WithQueue, babelqueue.WithTraceID); meta.queue records the logical BabelQueue queue, which need not equal machinery's broker queue.
Source Files