The Amazon SQS + SNS connector in redb.Route, at-least-once and pub/sub via SNS SQS. Still leaving MassTransit

series: "redb.Route"

We've done Kafka and RabbitMQ earlier in this series. Next up: Amazon SQS — and riding along in the same package, SNS. Two transports, because in AWS-land they travel together. The redb.Route.Sqs connector sits on the native AWS SDK for .NET v4 (AWSSDK.SQS, AWSSDK.SimpleNotificationService), but you don't write a "client" — you write routes, and the whole queue collapses into a single URI:

From("sqs://orders?waitTimeSeconds=20&concurrentConsumers=4")
    .Log("Order in: ${body}")
    .To("direct://process");

Read it once and the whole story's there: long-poll the orders queue in 20-second pulls, four consumers wide, log it, pass it on. No "new up a client," no ReceiveMessage, no "remember to DeleteMessage after you've handled it" — the connector does all of that.

About "leaving MassTransit," same disclaimer as last time: it's not that MassTransit can't do this. It's had SQS/SNS for years — MassTransit.AmazonSQS — and it wires up the same SNS→SQS pairing. The difference is the model. MassTransit is a bus — message contracts, consumers, bus config, the whole worldview bought at once. redb.Route is explicit routes in the Apache Camel spirit: an endpoint is a URI, the integration patterns are steps in a route, the transport is an abstraction underneath. This post is about what SQS and SNS look like in that second model.

And the thing everyone's actually asking about right now — worth stating precisely rather than gleefully. MassTransit went commercial with v9: the line now ships under Massient, Inc., and MassTransit.AmazonSQS 9.0.0 on NuGet is already from there. But no spin: v8 stays Apache 2.0, with security patches and critical fixes through at least the end of 2026; v9 isn't a closed binary — it's source-available; and organizations under $1M USD annual revenue (plus non-profits under $1M in expenses) get it at a 100% discount — free, minus commercial support. So "MassTransit is paid now" is false for a small team and true for a large one, where it turns into a real decision: pay, ride v8 to the end of its support, or look around. redb.Route is Apache 2.0, this SQS connector included — but pick it because the routing model fits your head, not because of a license line. A license is a reason to look around; it isn't an argument about architecture.

All the code is in English; the examples aren't invented — they run against LocalStack (a Docker container on http://localhost:4566), the same way the connector's integration tests do.

Part of the redb / redb.Route series — recent posts first:

• The RabbitMQ connector — the whole broker in one URI: RPC, competing consumers, dead-lettering (previous in the series)
• Leaving MassTransit for a Camel state of mind: the Kafka connector, Scatter-Gather, and transactions
• Apache Camel for .NET, dissected: the HTTP connector with no ASP.NET MVC + the Content-Based Router
• redb.Route — Apache Camel for .NET: 22 transports, 30+ EIP patterns, compiled DSL

Sources: github.com/redbase-app. About the database itself: redbase.app.

---

One package, two transports

redb.Route.Sqs registers two schemes, and they don't play the same role:

• sqs:// — a queue. Works as both a consumer (in From(...)) and a producer (in .To(...)).
• sns:// — a topic. Publisher-only. SNS has no pull consumer — delivery is push (to SQS / HTTP / email / SMS). Try to From("sns://...") and it throws NotSupportedException on purpose: to "read" a topic, you subscribe an SQS queue to it and read that with sqs://.

Why they share a package: at AWS they're a pair. SQS is a durable point-to-point queue (one message, one consumer). SNS is a publish-subscribe topic (one message, every subscriber). The canonical AWS fan-out is an SNS topic with several SQS queues hanging off it — so one connector covers both: sqs:// and sns://.

Wiring it into DI:

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteSqs();   // registers both sqs:// and sns://
    route.AddRouteBuilder<MyRoutes>();
});

---

Anatomy of the URI

An endpoint is a string:

sqs://queue-name?region=us-east-1&waitTimeSeconds=20&concurrentConsumers=4
sns://topic-name?region=us-east-1&autoCreateTopic=true

Scheme, the queue or topic name in the path, then query parameters (names are in the tables below). The fluent builder gives you the same thing: Sqs.Queue("orders").WaitTimeSeconds(20)... and Sns.Topic("events").Region("us-east-1")... build that exact string, URL-encoding included.

Three things are AWS-specific and worth a pause:

Don't repeat your credentials. Twenty queues on one account? Register an AwsConnectionFactory in the registry and point at it with connectionFactory=name. One factory builds both the SQS and the SNS client — they share credentials, region and service URL:

context.AddToRegistry("prod-aws", new AwsConnectionFactory
{
    Region = "eu-west-1",
    UseDefaultCredentialsProvider = true,   // IAM role on the instance / SSO / env vars
});

From("sqs://orders?connectionFactory=prod-aws&concurrentConsumers=4");
To("sns://events?connectionFactory=prod-aws");

LocalStack / ElasticMQ via serviceUrl=. An explicit URL overrides the regional endpoint (the region sticks around only to sign the request). That's your local-dev and test mode:

sqs://orders?serviceUrl=http://localhost:4566&region=us-east-1&accessKey=test&secretKey=test

Queues and topics get created on the spot. autoCreateQueue=true (or autoCreateTopic=true) creates the resource on startup if it's missing. A name ending in .fifo comes up as a FIFO resource automatically, with ContentBasedDeduplication=true.

---

Every parameter

The reason to bookmark this post. Names are exactly as they appear in the URI; defaults are what you get out of the box.

Connection and credentials (shared by sqs:// and sns://)

Parameter	Default	What it's for
region	us-east-1	AWS region
serviceUrl	(empty)	Explicit endpoint (LocalStack/ElasticMQ); overrides the region
accessKey / secretKey	(empty)	Static keys
sessionToken	(empty)	Temporary (STS) credentials
profileName	(empty)	A named profile from the shared credentials file
useDefaultCredentialsProvider	false	The default AWS chain (env / IAM role / SSO)
connectionFactory	(empty)	A factory in the registry; overrides URI credentials
retryCount	3	SDK retries on transient errors
connectionTimeout	30000	HTTP client timeout, ms
proxyHost / proxyPort	—	HTTP proxy

Resolution order: the default chain → a profile → a session token → static keys. Validate() insists on at least one path — accessKey+secretKey, profileName, useDefaultCredentialsProvider=true, or a registered connectionFactory — otherwise it throws at startup.

SQS — consumer (From("sqs://..."))

Parameter	Default	What it's for
waitTimeSeconds	20	Long-poll (0–20). Fewer empty receives, fewer API calls
maxNumberOfMessages	10	How many to pull per receive (1–10)
visibilityTimeout	0 (queue default)	How long a message stays hidden while you work it
concurrentConsumers	1	Number of competing receive loops (see the concurrency recipe)
extendMessageVisibility	false	Keep extending visibility while the handler runs (a heartbeat); needs visibilityTimeout > 0
deleteAfterRead	true	Delete the message after a clean pass (this is the ack)
resetVisibilityOnFailure	false	On error, reset visibility to 0 → immediate redelivery
transacted	false	Defer the delete into the route transaction (.Transacted())
attributeNames	All	Which system attributes to request
messageAttributeNames	All	Which message attributes to request
delay	0	Pause (ms) after an empty receive before polling again
initialDelay	0	Delay (ms) before the first poll

SQS — producer (.To("sqs://..."))

Parameter	Default	What it's for
delaySeconds	0	Delivery delay (0–900). FIFO queues reject a per-message delay
messageGroupId	—	FIFO group id. A constant or a ${...} expression (per message)
messageDeduplicationId	—	FIFO dedup id. Skip it if the queue has content-based dedup
enableBatch	false	Send an IEnumerable body as one SendMessageBatch
batchMaxMessages	10	Batch size (SQS's hard cap is 10)
autoCreateQueue	false	Create the queue on startup (FIFO if the name ends in .fifo)
queueUrl	(empty)	An explicit queue URL instead of resolving by name

SNS — publisher (.To("sns://..."))

Parameter	Default	What it's for
autoCreateTopic	false	Create the topic on startup (FIFO on .fifo)
topicArn	(empty)	An explicit topic ARN instead of resolving by name
subject	(empty)	Subject (for email delivery). Supports ${...}
messageStructure	(empty)	json — per-protocol payloads
messageGroupId / messageDeduplicationId	—	For FIFO topics
subscribeSnsToSqs + subscribeQueueArn	false	On startup, subscribe an SQS queue (by ARN) to this topic
rawMessageDelivery	false	On that subscription, deliver the bare payload (see the pub/sub recipe)

---

Headers and tracing through the broker

The sqs:// consumer stamps message metadata onto the headers (prefixed redbSqs.), and incoming message attributes land under redbSqs.attr.:

Header	What's in it
redbSqs.queue	The queue name
redbSqs.messageId	Message id
redbSqs.receiptHandle	Receipt handle (needed to delete or change visibility)
redbSqs.approximateReceiveCount	How many times this message has been delivered ("is this a retry?")
redbSqs.messageGroupId / redbSqs.sequenceNumber	FIFO group + sequence number
redbSqs.sentTimestamp	When it was sent (epoch millis)
redbSqs.attr.<name>	An incoming user message attribute

Going the other way, the header → message-attribute mapping on the producer goes like this: user headers ride out as attributes as-is; an incoming redbSqs.attr.<name> is forwarded under the bare <name> (the prefix is stripped, so sqs→sqs and sqs→sns bridges keep their attributes); the internal redbSqs.* / redbSns.* keys are dropped, so one hop's metadata doesn't leak into the next.

Tracing comes for free: before sending, the producer injects the W3C traceparent/tracestate as message attributes via the standard DistributedContextPropagator, and the consumer on the far end picks them up and continues the trace. Over plain SQS (sqs:// → sqs://) the chain is unbroken with zero setup. (There's a wrinkle when SNS is in the middle — see the pub/sub recipe.)

---

Straight talk on delivery: at-least-once, visibility, transacted

This is where the marketing likes to say "exactly-once." Let's read the code instead.

The SQS consumer is at-least-once. A message is deleted (DeleteMessage) only after it clears the route. The settle logic:

• clean pass + deleteAfterRead=true → DeleteMessage (the ack);
• failed + resetVisibilityOnFailure=true → ChangeMessageVisibility(0) → it comes right back (fast retry);
• failed without that flag → it simply times out on visibility and gets redelivered.

visibilityTimeout is how long the message is hidden while you work it. Long handler? There's extendMessageVisibility=true: a background heartbeat bumps the visibility every visibilityTimeout / 2, so slow work never triggers a redelivery out from under you.

transacted=true rolls the ack into the route transaction. The delete doesn't happen immediately — a deferred SqsAckAction is registered, whose Commit deletes the message and whose Rollback resets visibility to 0 (immediate redelivery). Commit and rollback ride along with your redb work at the .Transacted() boundary:

From("sqs://orders?transacted=true&visibilityTimeout=60")
    .Transacted()
        .ProcessWithRedb((redb, ex) => redb.SaveAsync(Parse(ex)))   // write to the DB
    .EndTransaction();     // DB committed → the message is deleted; DB failed → it comes back

What's not here is cross-queue "read from A, write to B" atomicity at the broker level — SQS doesn't have it to give. Crash in the gap between "work done" and "delete," and after a restart you get a repeat — at-least-once, not exactly-once. So make your handler idempotent; redbSqs.approximateReceiveCount is there to help. The honest label is "at-least-once with a transactional ack at the route level" — no exactly-once fairy dust.

Ordering holds only on a FIFO queue (.fifo) and only with concurrentConsumers=1.

---

Recipe 1: producer, consumer, FIFO, batch

The bare minimum:

// Consumer — long-poll, delete after success
From(Sqs.Queue("orders").WaitTimeSeconds(20))
    .Process(HandleOrder);

// Producer — send the body to the queue
To(Sqs.Queue("orders").Region("eu-west-1"));

FIFO — the name ends in .fifo, and a group id is required (a constant or a per-message expression):

To(Sqs.Queue("orders.fifo").MessageGroupId("${header.customerId}"));

Batch — an IEnumerable body goes out as a single SendMessageBatch (in chunks of batchMaxMessages, SQS's cap being 10):

From("timer://tick?period=5000")
    .SetBody(_ => new[] { "a", "b", "c", "d", "e" })
    .To(Sqs.Queue("orders").EnableBatch());   // one call instead of five

---

Recipe 2: competing consumers (concurrency)

One parameter — concurrentConsumers:

From(Sqs.Queue("orders").ConcurrentConsumers(8).MaxNumberOfMessages(1))
    .Process(HandleOrder);   // up to 8 messages handled at once

This is the SQS-native concurrency model: the connector spins up N independent receive loops (one task each), and each pulls and handles its own message. Setting maxNumberOfMessages=1 alongside it makes each of the N workers hold exactly one message at a time, so the pool saturates honestly instead of one loop grabbing a batch of 10. The default is 1 (strictly serial). Go N > 1 and per-queue ordering is off the table, and your handler had better be thread-safe.

How much this actually parallelizes shows up in the SqsRpcDemo (more on it below): 12 requests, roughly 300 ms of work each, a pool of 4 — the whole batch clears in about 1.2 s instead of the ~3.6 s a serial consumer would spend, and the measured peak concurrency pins at exactly 4.

And if what you want to parallelize is the processing inside the route rather than the intake (a serial source, heavy work), there's an orthogonal EIP step, .Threads(N). concurrentConsumers scales reading off the queue; .Threads(N) scales the work; the CONCURRENCY.md guide walks through the difference. They compose freely.

---

Recipe 3: RPC (request/reply) — and why SQS won't hand it to you

Here's the honest contrast with the last post. On RabbitMQ, RPC flipped on with one flag — replyTo=true on the client and zero config on the server, because the broker does reply queues itself. SQS has none of that — no protocol-level reply-to, no built-in waiting for an answer. RPC over SQS is something you build, using the classic correlation pattern: a reply queue, a correlationId, and matching the answer back. That's exactly what the SqsRpcDemo shows.

The client tags each request with two message attributes — correlationId and replyTo (the queue to answer on) — and drops it on the request queue. The worker computes a result and sends it to the queue named in replyTo; the correlationId rides back on its own (the incoming redbSqs.attr.correlationId is forwarded straight back out as an attribute). The worker is a plain consumer plus a dynamic .ToD(...):

// Worker: N competing consumers on the request queue, reply to the queue named in replyTo
From(Sqs.Queue("rpc-requests").ConcurrentConsumers(4).MaxNumberOfMessages(1))
    .Process(async (ex, ct) =>
    {
        var n = int.Parse(ex.In.Body!.ToString()!);
        await Task.Delay(300, ct);              // simulated work
        ex.In.Body = (n * (long)n).ToString();  // the reply body
    })
    // dynamic destination — the reply queue comes from the request's replyTo attribute
    .ToD(ex => Sqs.Queue(ReplyTo(ex)).Build());

On the client side a separate route reads the reply queue and matches the correlationId back to the pending call:

From(Sqs.Queue("rpc-replies").MaxNumberOfMessages(10))
    .Process((ex, ct) =>
    {
        var id = Attr(ex, "correlationId");
        if (id is not null && Pending.TryRemove(id, out var tcs))
            tcs.TrySetResult(ex.In.Body?.ToString() ?? "");
        return Task.CompletedTask;
    });

The takeaway: RPC on SQS works, but it's assembled from primitives (attributes + .ToD + correlation), not toggled with a flag. When you want RPC out of the box, you reach for a broker with a native reply path (RabbitMQ). SQS is honest about what it is: durable queues and redelivery, and request/reply is a thing you wire up on top. The full, runnable version lives in redb.Route/demos/SqsRpcDemo (it runs against LocalStack).

---

Recipe 4 (EIP): Publish-Subscribe via SNS→SQS fan-out

Here's the headline pattern, the whole reason SQS keeps SNS company. Publish-Subscribe Channel, straight out of Hohpe & Woolf: the publisher sends one message and every subscriber gets it — independently, into its own queue, with its own retries. At AWS you don't do this in code, you do it with topology: an SNS topic with several SQS queues subscribed to it.

publish "order" ──▶ SNS topic ──┬──▶ SQS "orders-billing"  ──▶ billing route
                                 └──▶ SQS "orders-shipping" ──▶ shipping route

The publisher just publishes to the topic, and every subscribed queue gets a copy — each with its own independent consumer:

// Publisher — publish an order event; SNS fans it out to every subscribed queue:
From("timer://orders?period=5000")
    .SetBody(_ => BuildOrder())
    .To(Sns.Topic("orders-events"));

// Two independent subscribers — each its own SQS queue, each its own consumer:
From(Sqs.Queue("orders-billing")).Process(Charge);
From(Sqs.Queue("orders-shipping")).Process(Ship);

The subscriptions themselves the connector can set up from the URI — subscribeSnsToSqs + the queue ARN; the subscribe runs when the sns:// producer starts. Each sns:// endpoint subscribes one queue (a single ARN), so a two-queue fan-out is two subscriptions on the one topic:

// Subscribe each queue to the topic with raw delivery (bare payload, no envelope):
Sns.Topic("orders-events").SubscribeSnsToSqs(billingArn).RawMessageDelivery();
Sns.Topic("orders-events").SubscribeSnsToSqs(shippingArn).RawMessageDelivery();

For the exact fan-out wiring (creating the queues, reading their ARNs, subscribing, publishing), see redb.Route/demos/SqsPubSubDemo.

⚠️ The catch: the envelope vs. raw

By default SNS does not drop your payload into the queue as-is — it wraps it in a JSON notification envelope:

{ "Type": "Notification", "MessageId": "...", "TopicArn": "...",
  "Message": "<your payload, as a string>", "MessageAttributes": { ... } }

So the subscribing queue receives the envelope, not your message, and your SNS attributes sit inside it rather than as SQS attributes. That's standard AWS SNS behavior (the subscription's Raw Message Delivery = OFF), and it earns its keep in exactly one case: when a single queue listens to several topics and you need the envelope's TopicArn to tell them apart. For an ordinary fan-out it's just in the way — you end up unwrapping the envelope, and the SNS→SQS trace breaks (the traceparent goes inside the envelope, while the SQS consumer looks for it among the SQS attributes).

The fix is the subscription's Raw Message Delivery = true: the queue gets the bare payload, and the SNS attributes become SQS attributes (and the trace is whole again). In the connector that's .RawMessageDelivery() — it sets RawMessageDelivery=true on that subscribeSnsToSqs subscription. The SqsPubSubDemo prints the first delivered body: with .RawMessageDelivery() it's {"orderId":1,"amount":100}, not an envelope.

On rawMessageDelivery. This option lands in the next NuGet release of redb.Route.Sqs. You can already grab it from source right now — the redb.Route.Sqs connector on GitHub (the connector folder — SnsEndpointOptions / SnsProducer / the RawMessageDelivery fluent method). Without it, SNS→SQS still works; you just get the JSON envelope and unwrap it yourself (envelope.Message).

---

Recipe 5 (a cousin): Multicast — the same fan-out, but inside the route

Publish-Subscribe copies the message on the broker (SNS hands a copy to each subscribed queue). There's a close cousin that copies inside the route — Multicast: your step sends a copy of the exchange to several destinations itself.

From("sqs://orders")
    .Multicast()
        .To("sqs://orders-billing")
        .To("sqs://orders-shipping")
    .End();

Looks similar, but the difference is fundamental — where the copying happens, and who knows about whom:

	SNS→SQS (pub/sub)	Multicast
Where it copies	on the broker (SNS)	in the route (your process)
Who knows the recipients	SNS (the subscriptions)	the route (the .To list)
Adding a recipient	subscribe another queue, leave the publisher alone	edit the route's code
Coupling	fully decoupled	recipients are baked into the route
Parallelism / aggregating replies	none (fire-and-forget per subscription)	yes: .Parallel().MaxParallelism(N), merge the branches

The rule of thumb: need to decouple the publisher from the subscribers (anyone can subscribe later, the publisher never knows) — SNS→SQS. Need to fan out to a fixed list right here and maybe collect the answers — Multicast (or its sibling Scatter-Gather from the Kafka post). One is broker topology, the other is a route step; you pick by where the fan-out logic should live.

---

Wrap-up

SQS and SNS in redb.Route are routes where the whole queue and the whole topic are one URI: From("sqs://…") / .To("sqs://…") / .To("sns://…"), with all the AWS SDK plumbing, the receive/delete dance, visibility, subscriptions and correlation tucked behind the string's parameters. Long-poll, FIFO, batching, competing consumers on a single parameter, at-least-once with a transactional ack, SNS→SQS fan-out with a choice of "envelope or bare payload" — the set is complete, and what you're left with is a short, legible string.

Straight talk on the trade-offs: there's no exactly-once (SQS itself doesn't offer it) — you get at-least-once plus idempotency; RPC isn't out of the box, it's assembled from attributes and .ToD; and an SNS subscription wraps your payload in an envelope by default. All of that is AWS being AWS, not the connector cutting corners — and the connector is upfront about every bit of it.

Two runnable examples — redb.Route/demos/SqsRpcDemo (RPC + concurrency) and redb.Route/demos/SqsPubSubDemo (SNS→SQS fan-out + raw delivery); both come up on LocalStack (http://localhost:4566, test/test) and run as a self-test. If something bites in your scenario, say so in the comments.

Sources and releases: github.com/redbase-app. About the redb database: redbase.app.

---

If this was useful — a ⭐ on GitHub helps others find it.