BabelQueue
27 GitHub Get Started

Runtime & transports

Last updated:

The Go core is a codec, but it also ships an optional, still zero-dependency App runtime plus a Transport interface and an in-memory transport. Broker drivers live in separate modules so the core stays dependency-free: …/redis (built on go-redis), …/amqp (built on amqp091-go), …/sqs (built on aws-sdk-go-v2), …/azureservicebus (built on azservicebus), …/pulsar (built on the pure-Go pulsar-client-go), …/kafka (built on the pure-Go segmentio/kafka-go) and …/artemis (Apache ActiveMQ Artemis over the pure-Go Azure/go-amqp, AMQP 1.0).

The App runtime

func NewApp(transport Transport, opts ...AppOption) *App

Options: WithDefaultQueue(string) (default "default"), WithMaxAttempts(int) (default 3), WithUnknownURNStrategy(string) (default StrategyFail), WithDeadLetter(bool), WithDeadLetterQueue(string), WithDeadLetterSuffix(string) (default ".dlq"; on the SQS transport the default is "-dlq" and an explicit suffix goes before .fifo), WithPollTimeout(time.Duration), WithRetryBackoff(time.Duration) (delay before a failed message is redelivered; default 0; applies to transports that implement Releaser, such as SQS), WithUnknownURNReleaseDelay(time.Duration) (redelivery delay for the StrategyRelease unknown-URN strategy; default 0; like WithRetryBackoff, only effective on Releaser transports), WithAckErrorHandler(fn) (observe a failed acknowledgement; the message is not retried because its ack failed) and WithReleaseErrorHandler(fn) (observe a failed in-place release; the broker redelivers once the reservation lapses).

Methods:

  • Handle(urn string, handler Handler) — register a handler. Handler is func(ctx, env Envelope) error; returning an error triggers retry/dead-letter, nil acks.
  • Publish(ctx, urn string, data map[string]any, opts ...Option) (string, error) — builds, encodes and publishes the canonical envelope; returns meta.id.
  • PublishWithHeaders(ctx, urn string, data map[string]any, headers map[string]string, opts ...Option) (string, error) — like Publish, plus out-of-band transport headers (e.g. a W3C traceparent) carried beside the envelope. Transports without header support fall back to a plain publish.
  • Consume(ctx, queue ...string) error — blocks, routing by URN until ctx is cancelled (one bad message never stops the loop). Run(ctx) is the alias on the default queue.
  • Drain(ctx, queue string, max int) (int, error) — process up to max (or until empty), returns the count. Handy for tests / one-shot workers.

Unknown-URN strategies are the string constants StrategyFail, StrategyDelete, StrategyRelease, StrategyDeadLetter.

In-memory example

package main

import (
	"context"
	"fmt"

	babelqueue "github.com/babelqueue/babelqueue-go"
)

func main() {
	ctx := context.Background()
	app := babelqueue.NewApp(babelqueue.NewInMemoryTransport(),
		babelqueue.WithDefaultQueue("orders"),
		babelqueue.WithDeadLetter(true),
	)

	app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
		fmt.Println(env.Data["order_id"], env.TraceID)
		return nil
	})

	app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
	app.Drain(ctx, "orders", 0) // Consume(ctx) blocks for a long-running worker
}

Redis transport

go get github.com/babelqueue/babelqueue-go/redis

Reliable-queue pattern (RPUSH to produce, a Lua script that issues LMOVE to a :processing list to reserve, LREM to ack; Redis 6.2+). Create it with a URL or your own client:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/redis"
)

tr, err := redis.New("redis://localhost:6379/0") // or redis.NewWithClient(client)
if err != nil {
	return 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 {
	return nil // handle env.Data
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx) // blocks

Options:

  • redis.WithProcessingSuffix(string) (default ":processing").
  • redis.WithRetryAfter(time.Duration) sets the recovery window (default 90 s); redis.WithoutRetryAfter() is retry_after: null (see the +inf note). The default layout sweeps expired :processing entries back to the ready list through <queue>:processing:deadlines; keep the handler timeout shorter than retry_after.
  • redis.WithLaravelLayout(prefix) switches to Laravel’s sorted-set layout (queues:<name>, :delayed, :reserved, :notify), where delays and recovery-as-an-attempt apply. The default layout has no delayed set.
  • redis.WithMigrationBatchSize(n) bounds one migrate or sweep call (-1, the default, moves all).

See Redis in the binding spec.

RabbitMQ (AMQP) transport

go get github.com/babelqueue/babelqueue-go/amqp

Durable queue, persistent delivery, basic.get + manual ack (at-least-once). The connection is lazy — it (re)connects on first use:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/amqp"
)

tr := amqp.New("amqp://guest:guest@localhost:5672/")
defer tr.Close()

app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	return nil
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx)

Each published message carries the contract AMQP properties so consumers can route without parsing the body: type = URN, correlationId = trace_id, messageId = meta.id, plus x-attempts / x-schema-version / x-source-lang headers, application/json + persistent delivery.

Amazon SQS transport

go get github.com/babelqueue/babelqueue-go/sqs

A babelqueue.Transport over the official AWS SDK (aws-sdk-go-v2): the canonical envelope is the MessageBody, projected onto native MessageAttributes; a visibility-timeout reserve → DeleteMessage ack gives at-least-once. A failed handler’s message is released with ChangeMessageVisibility (default delay 0 s, so it redelivers at once); pair each queue with a native RedrivePolicy (maxReceiveCount → a DLQ) so a message that always fails cannot loop forever. Build it with your AWS config (region from the standard chain) or your own client:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/sqs"
)

tr, err := sqs.New(ctx, sqs.WithRegion("eu-central-1")) // or sqs.NewWithClient(client)
if err != nil {
	return err
}

app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	return nil
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx) // blocks

Options: WithEndpoint (LocalStack/ElasticMQ), WithQueueURLPrefix (skip GetQueueUrl), WithWaitTimeSeconds, WithVisibilityTimeout, WithFIFO, WithMessageGroupID, WithContentDedup, WithClient. On SQS the dead-letter queue is <queue>-dlq (FIFO: <queue>-dlq.fifo), not <queue>.dlq: SQS does not accept a dot in a queue name, so the SQS transport (sqs v1.12.0) supplies that name as the default when you turn dead-lettering on. WithDeadLetterQueue names the queue outright; WithDeadLetterSuffix replaces -dlq, and on a FIFO queue the suffix is placed before .fifo too, because an SQS FIFO name must end in .fifo (orders.fifo + -x is orders-x.fifo, never orders.fifo-x). Each message carries the §3 MessageAttributes (bq-job = URN, bq-trace-id = trace_id, bq-message-id = meta.id, plus bq-schema-version / bq-source-lang / bq-created-at) so a consumer can route without decoding the body; attempts is reconciled to ApproximateReceiveCount − 1. See the SQS binding.

Azure Service Bus transport

go get github.com/babelqueue/babelqueue-go/azureservicebus

A babelqueue.Transport over the official Azure SDK (azservicebus): the canonical envelope is the message Body, projected onto native fields (Subject = URN, CorrelationID = trace_id, MessageID = meta.id) plus the bq- application properties; a PeekLock reserve → CompleteMessage ack gives at-least-once. Build it from a connection string, your own *azservicebus.Client, or an injected client:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/azureservicebus"
)

tr, err := azureservicebus.New(azureservicebus.WithConnectionString(connStr))
if err != nil {
	return err
}

app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	return nil
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx) // blocks

Options: WithAzureClient(*azservicebus.Client) (e.g. a namespace + token credential), WithClient, WithMaxWaitTime. attempts is reconciled to max(body.attempts, DeliveryCount − 1). Requires Go 1.25+. See the Azure Service Bus binding.

Apache Pulsar transport

go get github.com/babelqueue/babelqueue-go/pulsar

A babelqueue.Transport over the pure-Go apache/pulsar-client-go (no CGo, no libpulsar): the canonical envelope is the message payload, projected onto native Pulsar message properties (bq-job = URN, bq-trace-id = trace_id, bq-message-id = meta.id, plus bq-schema-version / bq-source-lang / bq-attempts, all string→string); a receive → Ack gives at-least-once. Build it from a service URL, your own pulsar.Client, or an injected client:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/pulsar"
)

tr, err := pulsar.New(pulsar.WithURL("pulsar://localhost:6650"))
if err != nil {
	return err
}

app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	return nil
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx) // blocks

Options: WithPulsarClient(pulsar.Client), WithClient, WithSubscription (default babelqueue), WithSubscriptionType (default Shared), WithTopicPrefix, WithMaxWaitTime. attempts is reconciled to max(body.attempts, RedeliveryCount) — the redelivery count is 0-based, so no −1. Requires Go 1.25+. See the Apache Pulsar binding.

Apache Kafka transport

go get github.com/babelqueue/babelqueue-go/kafka

A babelqueue.Transport over the pure-Go segmentio/kafka-go (no CGo, no librdkafka): the canonical envelope is the record value, projected onto native Kafka record headers (bq-job = URN, plus bq-trace-id / bq-message-id / bq-schema-version / bq-source-lang / bq-attempts, all UTF-8 byte strings) with the record timestamp mirroring meta.created_at. Consuming is process-then-commit: FetchMessage reserves a record (no auto-commit) and Ack commits the offset. Build it from brokers + a group id, or an injected client:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/kafka"
)

tr, err := kafka.New(kafka.WithBrokers("localhost:9092"), kafka.WithGroupID("orders-workers"))
if err != nil {
	return err
}

app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	return nil
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx) // blocks

Options: WithBrokers, WithGroupID, WithWriter / WithReaderFactory (for tests), WithMaxWaitTime. Kafka has no native delivery count, so attempts is reconciled from the authoritative bq-attempts header (else the body); the App owns retry by republishing with attempts+1 and dead-letters to <queue>.dlq. The retry, the dead-letter writes and babelqueue.Redrive (the write back to the work topic and every restore write back to the DLQ) keep the consumed record’s key (a null key stays null; see Record key below for empty keys). A restore re-writes the DLQ record as it was, with its original value bytes, headers and key, and a fresh timestamp. If a restore write fails, the consumer is rewound to that record before the run stops, so a later cumulative commit cannot skip it. Requires Go 1.24+. See the Apache Kafka binding.

Default Writer. When you do not pass WithWriter, the transport builds its own Writer with RequiredAcks: RequireAll and its own kafka-go Transport with a 1 s MetadataTTL. The write now waits for the in-sync replicas to acknowledge it, so a broker-side produce failure (for example in a retry or dead-letter republish) is returned instead of going unnoticed, and Publish latency includes that acknowledgement. The short metadata TTL avoids the 6 s cache of kafka-go’s shared default transport, which also caches a negative answer: a first write to a topic created moments earlier no longer fails with Unknown Topic Or Partition. A Writer injected with WithWriter is used as is.

(*Transport).Close() closes the readers the transport opened and, when it built its own Writer, that Writer and its private kafka-go transport; an injected Writer is left to its owner. After Close, Pop and Ack return ErrClosed.

Record key. The transport implements the optional KeyedRepublisher capability, so the retry republish, the envelope dead-letter write, the poison dead-letter write and the raw version-gated write carry the consumed record’s key as kafka-go’s Reader reports it. A null key stays null and a non-empty key is kept byte for byte; the key has no contract meaning and is never invented. Partition equality across SDKs is not guaranteed, because client partitioners differ.

Known limitation — empty keys. kafka-go’s high-level Reader returns a zero-length key as nil, so the Go consumer cannot tell an empty key from a null one. Its retry and dead-letter writes, and babelqueue.Redrive, therefore write an empty key back as null. The other SDKs keep an empty key empty. Broker-bindings §6.5 states key preservation as a SHOULD, so this is a known gap rather than a contract violation; avoid relying on empty-versus-null keys when a Go consumer is in the path.

Apache ActiveMQ Artemis transport

go get github.com/babelqueue/babelqueue-go/artemis

A babelqueue.Transport over the pure-Go Azure/go-amqp (AMQP 1.0, no CGo) — Artemis speaks AMQP 1.0, not RabbitMQ’s 0-9-1, so this is a distinct module from …/amqp. The canonical envelope is the message body, projected onto the AMQP a JMS peer reads: the x-opt-jms-type annotation = URN, correlation-id = trace_id, creation-time = meta.created_at, plus the string bq_ application properties (bq_schema_version / bq_source_lang / bq_attempts / bq_app_id; underscored, since JMS property names must be valid Java identifiers). Consuming is Receive → process → Ack (accept) for at-least-once. Build it from an AMQP URL or an injected client:

import (
	babelqueue "github.com/babelqueue/babelqueue-go"
	"github.com/babelqueue/babelqueue-go/artemis"
)

tr, err := artemis.New(ctx, artemis.WithURL("amqp://localhost:5672"))
if err != nil {
	return err
}

app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	return nil
})

app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
return app.Consume(ctx) // blocks

Options: WithURL (amqp:// / amqps://), WithClient (for tests), WithMaxWaitTime. attempts is reconciled to max(body, delivery-count) — the AMQP delivery-count is 0-based, so no −1 — and the App owns retry by republishing with attempts+1, dead-lettering to <queue>.dlq. Requires Go 1.25+. See the Apache ActiveMQ Artemis binding.

Retry, attempts & dead-letter

On a handler error the top-level attempts counter is incremented and the message is re-published to the same queue until WithMaxAttempts is reached; after that, if WithDeadLetter(true) is set, it is routed to the dead-letter queue (default: the source queue + .dlq) with a dead_letter block — otherwise it is dropped. See the wire contract.

If the dead-letter write itself fails (for example the DLQ is unreachable), the message is not acknowledged and is not lost: the App reports the error to WithDeadLetterErrorHandler, and on transports that support Rewind (Kafka, Pulsar, Azure Service Bus, Artemis) hands the delivery back, waiting WithRewindPause (default 1 s) before the broker redelivers it. On a redelivery the handler runs again, so keep handlers idempotent (see the idempotency helper).

If the re-publish for a retry (a handler error or StrategyFail) or for a StrategyRelease unknown URN fails, the message is not acknowledged, like a failed dead-letter write. This applies to a transport that cannot release a message in place, so the App falls back to re-publish and ack. The App logs the failure, reports it to WithReleaseErrorHandler and hands the message back: through Rewind where the transport supports it, otherwise it stays reserved and is redelivered under that transport’s reservation rules. The handler may run again, so keep handlers idempotent.

Undecodable (poison) bodies

With dead-lettering on (WithDeadLetter(true) or the StrategyDeadLetter strategy), a body that cannot be decoded as an envelope is never retried, because a retry cannot change the outcome. It is written straight to <queue>.dlq and acknowledged. The record follows ADR-0035: a body that is a strict JSON object (without a top-level dead_letter member) is spliced byte for byte, so every original byte is kept and only the dead_letter member is inserted; any other body (not JSON, a list, a scalar, trailing data, empty) is wrapped as {"raw": <body>}. Either way the record gets a dead_letter block with reason: "poison", the decode error in error, original_queue, lang and failed_at. Envelope keys are matched exactly as spelled: a member such as JOB or Data is an unknown member, not the job or data field, so it is kept as it is and the body is not poison on that account; when a key is repeated, the last one wins (the values are not merged). A body that is not valid UTF-8 cannot be represented in a JSON string, so it is carried as the standard base64 of its exact bytes in raw, with "raw_encoding": "base64"; invalid bytes are never replaced with U+FFFD. As above, the delivery is acknowledged only after the dead-letter write succeeds; if the write fails it stays unacknowledged and is handed back. With StrategyDelete the poison body is acknowledged and dropped. With dead-lettering off (any strategy but StrategyDelete, including the default StrategyFail), none of this applies: the body is retried up to WithMaxAttempts and then dropped. A transport that can release in place leaves the original body untouched; on any other transport the retry is re-published as an empty envelope, so the original body is lost. The SQS and Azure Service Bus transports are the exception (ADR-0037, since sqs v1.12.0, azureservicebus v1.3.0, core v1.18.0): with dead-lettering off (and not StrategyDelete) a poison body is not retried and not dropped on the first delivery, it is handed to the broker’s own mechanism. The SQS transport leaves the message undeleted, so visibility expiry redelivers it and the queue’s RedrivePolicy moves it to its DLQ; the Azure Service Bus transport abandons it until MaxDeliveryCount moves it to the native dead-letter sub-queue, or dead-letters it there explicitly.

The Redis transport ignores the two opt-outs above: it flags a poison delivery itself and dead-letters it in every configuration, so StrategyDelete and dead-lettering being off make no difference there, and the poison body is never retried (broker-bindings §1.1).

Schema-version gate

The SQS, Azure Service Bus, Kafka and Artemis transports check the bq-schema-version message attribute / property / header (Artemis: bq_schema_version) before the body is decoded. A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. A property that is present with a null value counts as missing too (Kafka, Artemis, Azure Service Bus). Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body. On Azure Service Bus the value may also be an integer 1 of any width; a float (including 1.0), boolean or byte value is rejected. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation.

A rejected delivery reaches the App on a reserved route, babelqueue:unsupported-schema-version:<version>. The transport gates without decoding the body, and the App does not decode it either. In the route, <version> is URL-escaped and cut to 64 runes plus .... No handler can match that route and a retry cannot change the outcome, so the App does not retry it. The StrategyDelete unknown-URN strategy wins over dead-lettering: the delivery is acknowledged and dropped. Otherwise, with WithDeadLetter(true) (or the StrategyDeadLetter strategy) the incoming body is written to the dead-letter queue byte-identical, with no dead_letter block and no unknown_urn reason, then acknowledged. With dead-lettering off it is acknowledged and dropped on Kafka and Artemis, while the SQS and Azure Service Bus transports hand it to the broker’s own mechanism instead (SQS: left undeleted for the RedrivePolicy; Azure Service Bus: abandoned or dead-lettered natively), so nothing is dropped there (ADR-0037, since sqs v1.12.0, azureservicebus v1.3.0, core v1.18.0). If the dead-letter write fails the message is not acknowledged: the App calls Rewind on transports that support it (Kafka, Artemis, Azure Service Bus), reports the error to WithDeadLetterErrorHandler, and waits WithRewindPause (default 1 s) so the broker redelivers it. Each of the four transports also takes an optional WithOnSchemaVersionRejected(func(version, queue string)) diagnostic hook.