Amazon SQS transport

Last updated:

BabelQueue.Sqs is an Amazon SQS transport on the .NET core. It sends the canonical envelope as the MessageBody with the §3 SQS MessageAttributes, 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.

Install

dotnet add package BabelQueue.Sqs

Requirements: .NET 8. It pulls BabelQueue.Core and the AWS SDK for .NET (AWSSDK.SQS) transitively. You supply an IAmazonSQS client, so its credentials/region come from your standard AWS configuration.

Produce

using Amazon.SQS;
using BabelQueue.Sqs;

IAmazonSQS sqs = new AmazonSQSClient(); // your AWS config / credentials
var url = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders";

var publisher = new SqsPublisher(sqs, url);
string id = await publisher.PublishAsync(
    "urn:babel:orders:created",
    new Dictionary<string, object?> { ["order_id"] = 1042 });

PublishAsync returns the message meta.id; pass a traceId to continue an existing trace. FIFO queues: new SqsPublisher(sqs, url, fifo: true) (the queue URL must end in .fifo).

Consume

Map URNs to handlers; PollAsync receives one batch, routes each message, and deletes the ones handled. A throwing handler’s message is released, never deleted: the consumer calls ChangeMessageVisibility with the configured release delay (default 0 s, so SQS redelivers it at once, at-least-once). Give the queue a native RedrivePolicy (maxReceiveCount → a DLQ) so a message that always fails cannot loop forever:

using Amazon.SQS;
using BabelQueue;
using BabelQueue.Sqs;

var handlers = new Dictionary<string, BabelHandler>
{
    ["urn:babel:orders:created"] = (envelope, message, ct) =>
    {
        var orderId = envelope.Data["order_id"];
        // ... use envelope.TraceId, envelope.Data
        return Task.CompletedTask;
    },
};

var consumer = new SqsConsumer(sqs, url, handlers, new SqsConsumerOptions
{
    OnUnknownUrn = (envelope, message, ct) => Task.CompletedTask, // no handler for this URN
    OnError = (error, envelope, message) => { /* loop never stops */ },
});

while (running)
{
    await consumer.PollAsync();
}

A handler is BabelHandler — delegate Task BabelHandler(Envelope, Message, CancellationToken). attempts is reconciled to ApproximateReceiveCount − 1 for the handler (never lowering a runtime-incremented count). Point the AmazonSQSClient’s endpoint at LocalStack/ElasticMQ for local testing.

Poison bodies and version-gated messages

A poison body (not JSON, not an envelope, no valid job or urn, or an unsupported meta.schema_version) and a version-gated message can never succeed, so the consumer settles them on their first delivery, never hands them to a handler and never retries them (ADR-0037, since 1.4.0).

Before decoding, the consumer version-gates on the bq-schema-version message attribute (§3.7). 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. 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, and OnError is called with an empty envelope. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation, and a body that fails it is a poison body.

What happens next depends on whether you set SqsConsumerOptions.DeadLetterQueueUrl:

var consumer = new SqsConsumer(sqs, url, handlers, new SqsConsumerOptions
{
    DeadLetterQueueUrl = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders-dlq",
});
Configuration Poison body Version-gated message
DeadLetterQueueUrl set sent to <queue>-dlq as the poison record (a strict JSON object body spliced with a dead_letter block, reason: "poison"; anything else wrapped under raw), then deleted the raw body, byte for byte and with no dead_letter block, sent to <queue>.dlq, then deleted
UnknownUrnStrategy = delete deleted, no record deleted, no record (dropped)
no DeadLetterQueueUrl reported and left to SQS: neither deleted nor released, so visibility expiry redelivers it and the queue’s RedrivePolicy moves it to its DLQ the same

The original is deleted only after the DLQ write succeeded; a failed write leaves it undeleted and reports it. Name the DLQ <queue>-dlq (SQS does not accept a dot in a queue name, so not <queue>.dlq; a FIFO one is <queue>-dlq.fifo). A FIFO DLQ gets MessageGroupId = the queue name and MessageDeduplicationId = the SQS message id. DeadLetterQueueUrl is opt-in: with none, nothing changes and the native RedrivePolicy stays your safety net. With one set, leaving a poison or version-gated message to the RedrivePolicy is no longer a conformant settle.

Only the UnknownUrnStrategy you set is read here (default fail; release behaves as fail). The OnUnknownUrn hook, and its precedence over the strategy, apply to an unknown URN only, so a hook-only consumer never drops a poison or version-gated message silently.

Because the wire format is the canonical envelope, the message a handler receives may have been produced by any BabelQueue SDK.