README
¶
babelqueue-go/asynq
asynq adapter for the BabelQueue Go runtime — "Polyglot Queues, Simplified."
asynq is the popular Redis-backed Go task queue. This adapter makes an asynq app produce and consume the canonical BabelQueue envelope, so it interoperates byte-for-byte with the PHP/Laravel, Python, … SDKs over Redis. asynq is not a new broker binding (its storage is Redis, §1) — it is a framework adapter, mirroring how Node's BullMQ adapter wraps the same core.
The mapping is direct and lossless:
| BabelQueue | asynq |
|---|---|
| canonical envelope JSON | task Payload (byte-identical — no wrapping, no added fields) |
job (the URN) |
task TypeName (asynq routes on it, like every BabelQueue consumer) |
Install
go get github.com/babelqueue/babelqueue-go/asynq
Requires Go 1.26+.
Produce
import (
babelqueue "github.com/babelqueue/babelqueue-go"
bqasynq "github.com/babelqueue/babelqueue-go/asynq"
"github.com/hibiken/asynq"
)
client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
defer client.Close()
p := bqasynq.NewProducer(client)
// Envelope options (WithQueue, WithTraceID) and asynq task options (MaxRetry,
// ProcessIn, …) are mixed freely — they're distinguished by type.
p.Dispatch(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042},
babelqueue.WithQueue("orders"), // recorded in meta.queue
asynq.MaxRetry(5), // forwarded onto the asynq task
)
meta.queue records the logical BabelQueue queue; asynq's own queue is set via
asynq.Queue(...) (default "default") and need not equal it.
Consume — route by URN
mux := asynq.NewServeMux()
bqasynq.Register(mux, "urn:babel:orders:created",
func(ctx context.Context, env babelqueue.Envelope) error {
// env.Data, env.TraceID … (trace_id preserved across the hop)
return nil
})
srv := asynq.NewServer(asynq.RedisClientOpt{Addr: "localhost:6379"}, asynq.Config{})
srv.Run(mux)
Register decodes the task payload into an envelope, rejects one that fails
consumer-side validation (ErrNotAccepted, so asynq applies its own retry/archive
policy), and routes by URN — the asynq-native equivalent of BabelQueue's URN
routing. For finer control, Bind(handler) returns a plain asynq.HandlerFunc you
can wire yourself, and Task/Envelope convert between an envelope and an
*asynq.Task directly.
How it maps to the contract
The envelope is the byte-identical payload (Redis §1
payload identity); the URN is the asynq TypeName. The envelope is unchanged
(schema_version stays 1); asynq is purely additive.
Test
go test ./... # unit + conformance (fake Enqueuer seam — no Redis)
The asynq client is reached through a small Enqueuer seam, so the suite runs
network-free.
License
MIT
Documentation
¶
Overview ¶
Package asynq bridges the github.com/hibiken/asynq task queue to the BabelQueue canonical envelope, so an asynq-based Go app interoperates byte-for-byte with the PHP/Laravel, Python, ... SDKs over Redis.
asynq is not a BabelQueue broker binding (its Redis storage is §1); it is a framework adapter, mirroring how Node's BullMQ adapter wraps the same core. The mapping is direct and lossless:
- the canonical envelope JSON IS the asynq task Payload (byte-identical to every other SDK — no wrapping, no added fields), and
- the asynq task TypeName IS the envelope URN (job), so asynq routes on the URN the same way every BabelQueue consumer does.
Produce wraps babelqueue.Make/Encode into an *asynq.Task and enqueues it; consume decodes the task Payload back into an babelqueue.Envelope and dispatches by URN through an *asynq.ServeMux.
// produce
p := asynq.NewProducer(client) // client is *hibiken/asynq.Client
p.Dispatch(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042},
babelqueue.WithQueue("orders"))
// consume — route by URN on an asynq ServeMux
mux := hasynq.NewServeMux()
asynq.Register(mux, "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); asynq is purely additive.
Full spec: https://babelqueue.com
Index ¶
- Variables
- func Bind(handler babelqueue.Handler) hasynq.HandlerFunc
- func Envelope(task *hasynq.Task) (babelqueue.Envelope, error)
- func Register(mux muxRegistrar, urn string, handler babelqueue.Handler)
- func Task(env babelqueue.Envelope, opts ...hasynq.Option) (*hasynq.Task, error)
- func TaskFor(urn string, data map[string]any, opts ...any) (*hasynq.Task, error)
- type Enqueuer
- type Producer
Constants ¶
This section is empty.
Variables ¶
var ErrNotAccepted = errors.New("babelqueue/asynq: 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 ¶
func Bind(handler babelqueue.Handler) hasynq.HandlerFunc
Bind adapts a BabelQueue babelqueue.Handler to an asynq handler: it decodes the task Payload into an envelope, rejects it with ErrNotAccepted when it fails babelqueue.Envelope.Accepts (so asynq applies its own retry/archive policy), and otherwise invokes handler. Use Register to also route by URN on a ServeMux.
func Envelope ¶
func Envelope(task *hasynq.Task) (babelqueue.Envelope, error)
Envelope decodes an asynq task's Payload into a BabelQueue envelope, resolving the urn alias (see babelqueue.Decode). The task TypeName is the URN, but the authoritative identity is the decoded body — TypeName is the routing key.
func Register ¶
func Register(mux muxRegistrar, urn string, handler babelqueue.Handler)
Register wires handler onto mux for the given URN: it registers Bind(handler) under the URN as the asynq TypeName pattern, so asynq routes tasks whose TypeName equals the URN to it — the asynq-native equivalent of BabelQueue's URN routing.
func Task ¶
Task builds an asynq task from an already-built envelope: TypeName = the URN (env.Job) and Payload = the canonical envelope JSON. Extra asynq options (e.g. hasynq.MaxRetry, hasynq.ProcessIn) are forwarded onto the task. It returns babelqueue.ErrEmptyURN when the envelope has no URN, and surfaces an encode error otherwise.
func TaskFor ¶
TaskFor builds the canonical envelope for (urn, data) via babelqueue.Make and wraps it as an asynq task (see Task). Pass envelope options (babelqueue.WithQueue, babelqueue.WithTraceID) and/or asynq task options — the two option families are distinguished by type. It mints a fresh trace id unless babelqueue.WithTraceID is given.
Types ¶
type Enqueuer ¶
type Enqueuer interface {
EnqueueContext(ctx context.Context, task *hasynq.Task, opts ...hasynq.Option) (*hasynq.TaskInfo, error)
}
Enqueuer is the subset of *hibiken/asynq.Client the Producer uses. The concrete client satisfies it, and a fake satisfies it in tests.
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer publishes BabelQueue envelopes as asynq tasks over an Enqueuer. It is safe for concurrent use when the underlying *asynq.Client is.
func NewProducer ¶
NewProducer wraps an Enqueuer (a *hibiken/asynq.Client, or a fake in tests).
func (*Producer) Dispatch ¶
func (p *Producer) Dispatch(ctx context.Context, urn string, data map[string]any, opts ...any) (*hasynq.TaskInfo, error)
Dispatch builds the canonical envelope for (urn, data) and enqueues it as an asynq task routed by URN, returning the asynq TaskInfo. Mix envelope options (babelqueue.WithQueue, babelqueue.WithTraceID) and asynq task options in opts; they are distinguished by type. The asynq queue defaults to the task's queue option (or asynq's "default") — pass hasynq.Queue(...) to set it, since meta.queue records the logical BabelQueue queue, which need not equal asynq's.
Source Files