README
¶
BabelQueue Redis transport (Go)
A Redis-backed babelqueue.Transport for the BabelQueue Go runtime,
using the reliable-queue pattern (RPUSH to produce, BLMOVE head → per-queue
processing list to reserve, LREM to ack). Built on
go-redis.
go get github.com/babelqueue/babelqueue-go/redis
import (
babelqueue "github.com/babelqueue/babelqueue-go"
"github.com/babelqueue/babelqueue-go/redis"
)
tr, err := redis.New("redis://localhost:6379/0")
if err != nil { panic(err) }
defer tr.Close()
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
// handle env.Data ...
return nil
})
app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
app.Consume(ctx)
Bring your own client with redis.NewWithClient(client) (custom pool / TLS / auth).
The BabelQueue core stays zero-dependency; this module is the only place go-redis
is pulled in.
Out-of-band headers (traceparent, ADR-0028)
The transport implements the optional babelqueue.HeaderPublisher capability, so
out-of-band transport headers — e.g. a W3C traceparent for cross-hop span linkage —
ride beside the frozen wire envelope, never inside it. Redis stores only the raw
list value (the LREM ack handle is that value), so there is no native per-message
metadata channel; the transport instead owns a small JSON frame, distinct from the
wire envelope:
{"__bq_frame":1,"headers":{"traceparent":"00-…-01"},"body":"<raw wire envelope>"}
Publishstores the bare envelope byte-for-byte (no frame).PublishWithHeaderswrites a frame only when the header map is non-empty; with no usable headers it degrades to a byte-identical barePublish.Popdetects frame-vs-bare by the reserved__bq_framesentinel (the frozen envelope never carries it). A frame is unwrapped —Bodyis the raw envelope,Headerscarries the frame's headers; theHandlestays the stored frame, so the reliable-queueRPUSH/BLMOVE/LREMsemantics are untouched. A bare value (an older or non-otel publisher, or a cross-version queue) consumes exactly as before:Headersis nil and the handle is the bare value.
Full spec: babelqueue.com · MIT © Muhammet Şafak
Documentation
¶
Overview ¶
Package redis is a Redis-backed babelqueue.Transport for the BabelQueue Go runtime. It uses the reliable-queue pattern: RPUSH to produce, BLMOVE the head to a per-queue processing list to reserve (so an in-flight message survives a worker crash), and LREM from that list to acknowledge.
tr, _ := redis.New("redis://localhost:6379/0")
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
This default layout is a Go-owned reliable queue. For a *shared* queue with Laravel workers, WithLaravelLayout switches the key names and the reservation to the broker-bindings.md §1.1 layout (see "Laravel-compatible layout" below).
It also implements the optional babelqueue.HeaderPublisher capability: out-of-band transport headers (e.g. a W3C traceparent for cross-hop span linkage, ADR-0028, or the bq-replay-bypass marker, ADR-0027) ride beside the frozen envelope (GR-1) — never in it — and are surfaced back to the consumer on babelqueue.ReceivedMessage.Headers.
Redis stores only the raw list value (the LREM ack handle *is* that value), so — unlike AMQP headers or SQS MessageAttributes — there is no native per-message metadata channel. To carry headers the transport owns a tiny JSON *frame* distinct from the wire envelope:
{"__bq_frame":1,"headers":{"traceparent":"00-..."},"body":"<raw wire envelope>"}
RPUSH stores the frame, so the LREM ack handle stays byte-for-byte what was pushed and the reliable-queue semantics (RPUSH/BLMOVE/LREM/processing list) are untouched. Framing is opt-in and backward compatible: only PublishWithHeaders with a non-empty map writes a frame; plain Publish (and PublishWithHeaders with no headers) stores the bare envelope byte-for-byte, exactly as before. Pop detects frame-vs-bare by the reserved "__bq_frame" sentinel — a frozen wire envelope can never carry it — and a bare value consumes with Headers=nil and Handle=value, so cross-version queues interoperate.
Laravel-compatible layout ¶
WithLaravelLayout puts the transport on the key names of broker-bindings.md §1.1, `<prefix><name>` (ready list), `:delayed` and `:reserved` (sorted sets) and `:notify` (blocking-pop wake-up), with `queues:` as the Laravel default prefix. Transport.Keys returns them. Publish pushes to the ready list and signals the notify list; Pop first moves due delayed and expired reserved members back to the ready list, then reserves the head into the reserved sorted set scored by its retry-after deadline; Ack removes it from that set. The reserved value is the stored value byte-for-byte, so payload identity (§1.2) holds. Not implemented: incrementing payload.attempts in Redis (the Laravel reserve script does that with cjson; the BabelQueue App counts attempts itself) and an API to schedule a delayed publish (the available-at formula is not defined by the contract).
Index ¶
- Constants
- type Keys
- type Option
- type Transport
- func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
- func (t *Transport) Close() error
- func (t *Transport) Keys(queue string) Keys
- func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
- func (t *Transport) Publish(ctx context.Context, queue, body string) error
- func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
Constants ¶
const DefaultKeyPrefix = "queues:"
DefaultKeyPrefix is Laravel's default Redis queue key prefix (`queues:`).
const DefaultRetryAfter = 90 * time.Second
DefaultRetryAfter is the reservation window of the Laravel layout (Laravel's `retry_after` default, 90 s): a reserved member whose deadline has passed is made ready again.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Keys ¶ added in v1.2.0
type Keys struct {
Ready string // list
Delayed string // sorted set, scored by available-at
Reserved string // sorted set, scored by the retry-after deadline
Notify string // list, blocking-pop wake-up
}
Keys are the Redis key names of one logical queue in the Laravel-compatible layout (broker-bindings.md §1.1).
func LaravelKeys ¶ added in v1.2.0
LaravelKeys returns the §1.1 key names for queue under prefix.
type Option ¶
type Option func(*Transport)
Option customizes New / NewWithClient.
func WithLaravelLayout ¶ added in v1.2.0
WithLaravelLayout switches the transport to the Laravel-compatible key layout of broker-bindings.md §1.1 under prefix (empty means DefaultKeyPrefix). The default layout (a `<queue>` list plus a `<queue>:processing` list) is unchanged when this option is not used.
func WithProcessingSuffix ¶
WithProcessingSuffix overrides the per-queue processing-list suffix (default ":processing").
func WithRetryAfter ¶ added in v1.2.0
WithRetryAfter sets the reservation window of the Laravel layout (default DefaultRetryAfter). It has no effect on the default layout.
type Transport ¶
type Transport struct {
// contains filtered or unexported fields
}
Transport implements babelqueue.Transport over Redis.
func NewWithClient ¶
NewWithClient wraps an existing go-redis client (bring your own pool/TLS/auth).
func (*Transport) Ack ¶
func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Ack removes the reserved message from its processing list (LREM). The handle is the raw stored list value (frame or bare body), so the LREM matches what was RPUSHed.
func (*Transport) Keys ¶ added in v1.2.0
Keys returns the key names the transport uses for queue. In the Laravel layout (WithLaravelLayout) these are the §1.1 names; in the default layout Ready is the queue list itself, Reserved is its processing list, and Delayed and Notify are empty (unused).
func (*Transport) Pop ¶
func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
Pop atomically moves the head of queue to its processing list and returns it, blocking up to timeout. It returns (nil, nil) when no message arrives in time.
The reserved value may be a header frame (written by PublishWithHeaders) or a bare wire envelope (written by Publish or an older/cross-version publisher). Either way the returned Handle is the *stored* value byte-for-byte, so Ack's LREM still matches.
func (*Transport) Publish ¶
Publish appends body to queue (RPUSH). The bare wire envelope is stored byte-for-byte (no frame), so a non-otel publisher and a cross-version consumer interoperate exactly as before.
func (*Transport) PublishWithHeaders ¶ added in v1.1.0
func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
PublishWithHeaders appends body together with out-of-band transport headers (babelqueue.HeaderPublisher). When the header map is non-empty the value RPUSHed is a transport-owned frame ({"__bq_frame":1,"headers":…,"body":<raw envelope>}) that carries the headers beside the frozen envelope (GR-1) — e.g. a W3C traceparent for cross-hop span linkage (ADR-0028). Blank keys/values are dropped; if nothing survives (or the map is nil/empty) it degrades to a byte-identical bare Transport.Publish, so nothing regresses.
Source Files