Redis transport
Last updated:
BabelQueue.Redis is a Redis transport on the .NET core, built on
StackExchange.Redis. It sends the
canonical envelope as a Redis list element with the
§1 reliable-queue pattern, and consumes by routing each
message to a handler by URN — so a message it produces is consumed by any other BabelQueue SDK,
and vice-versa.
Unlike the SQS or RabbitMQ bindings, a Redis list element carries no native metadata, so there
is no header/attribute projection: by default the list element is the canonical envelope
JSON, byte-for-byte, with no wrapping and no added fields. A consumer in any language pops that same
body and decodes it. When headers are published (PublishWithHeadersAsync, e.g. a W3C traceparent
per ADR-0028), the element is instead a __bq_frame
wrapper {"__bq_frame":1,"headers":…,"body":<envelope>} that every consumer unwraps; the envelope
inside is still unchanged.
Install
dotnet add package BabelQueue.Redis
Requirements: .NET 8. It pulls BabelQueue.Core and StackExchange.Redis transitively.
The consumer reserves with a Lua script that issues LMOVE, so it needs Redis 6.2+. On Redis Cluster, use
hash-tagged queue names; see Redis compatibility.
Produce
using BabelQueue.Redis;
using StackExchange.Redis;
var redis = await ConnectionMultiplexer.ConnectAsync("localhost:6379");
IDatabase db = redis.GetDatabase();
var id = await new RedisPublisher(db, "orders")
.PublishAsync("urn:babel:orders:created", new Dictionary<string, object?> { ["order_id"] = 1042 });
PublishAsync(urn, data) returns the message meta.id; pass a traceId to continue a trace.
Produce is RPUSH <queue> <envelope>.
Consume
using BabelQueue;
using BabelQueue.Redis;
var handlers = new Dictionary<string, BabelHandler>
{
["urn:babel:orders:created"] = async (env, rawBody, headers, ct) =>
{
// env.Data, env.TraceId, env.Attempts ...
},
};
var consumer = new RedisConsumer(db, "orders", handlers, new RedisConsumerOptions
{
OnError = (err, env, raw) => Console.Error.WriteLine(err),
});
await consumer.RunAsync(cancellationToken); // polls until cancelled
A reserve atomically moves the head of the queue onto a per-queue <queue>:processing list
(a Lua script that issues LMOVE <queue> <queue>:processing LEFT RIGHT), so an in-flight message
survives a crash; a successful handler LREMs it. Retry is at-least-once: a throwing handler
leaves the element on the processing list until the recovery sweep requeues it after RetryAfter. The poll loop never stops on a bad message — observe via OnError /
OnUnknownUrn. An unknown URN is passed to OnUnknownUrn (then acked) if one is set; otherwise
it is reported to OnError and dead-lettered to <queue>.dlq with dead_letter.reason: "unknown_urn"
(a fixed dead_letter, a documented deviation from the generic fail default). The DLQ write is unconditional: there is no DLQ opt-in switch, so it never degrades to delete.
Recovery and the zset layout
The default (RedisLayout.Legacy) layout writes a deadline member to
<queue>:processing:deadlines on every reserve and sweeps expired ones back to the ready list
before the next reserve. The window is RedisConsumerOptions.RetryAfter, 90 seconds by default;
null is retry_after: null (see the
+inf note). Keep the handler
timeout shorter than RetryAfter. A legacy recovery does not increment attempts, and a
ReleaseDelay is not available on the legacy layout.
For Laravel’s sorted-set layout, set Layout = RedisLayout.Zset on RedisConsumerOptions and
RedisPublisherOptions. It adds :delayed, :reserved and :notify, PublishDelayedAsync, and
recovery that counts as an attempt. See Redis in the binding spec.
Contract mapping (§1)
| Envelope | Redis |
|---|---|
| body | the list element (byte-identical across SDKs, no wrapping; a __bq_frame wrapper only when PublishWithHeadersAsync carries headers) |
job (URN) |
read from the decoded body (no native metadata to route on) |
trace_id / meta.id / … |
read from the decoded body |
| produce | RPUSH <queue> <envelope> |
| reserve | Lua script: LMOVE <queue> <queue>:processing LEFT RIGHT plus a deadline member (an in-flight message survives a crash) |
| ack | LREM <queue>:processing 1 <envelope> |
This is a .NET-owned reliable queue, mirroring the Go reference: produce/reserve/ack are
self-consistent and crash-safe on a queue this SDK owns end-to-end. Full parity with Laravel’s
reserved-sorted-set reservation on a shared PHP+.NET Redis queue is a separate task — see
broker-bindings §1.4. IDatabase is an interface, so the publisher and consumer unit-test against a
mock with no Redis and no network. The envelope is unchanged (schema_version stays 1); Redis
support is purely additive.