SQS — Basic Pubsub

Use this as your starting point for SQS integration: it covers the three non-sequenced topology shapes (minimal queue, DLQ, escalating retry backoff), all three SQS-compatible Outcome variants, and all three publish variants against a LocalStack backend. The SQS counterpart to the RabbitMQ basic example — same programming model, different infrastructure.

Prerequisites

Docker with a running daemon
LOCALSTACK_AUTH_TOKEN environment variable set to a valid LocalStack Pro token (LocalStack Pro is required for SNS+SQS FIFO queues)
Cargo feature: aws-sns-sqs

export LOCALSTACK_AUTH_TOKEN=<your-token>

A LocalStack testcontainer is started automatically — no manual docker run needed.

Run

LOCALSTACK_AUTH_TOKEN=<token> cargo run --example sqs_basic_pubsub --features aws-sns-sqs

Expected output

topologies declared

messages published

[minimal] order=ORD-001 amount=$50.00 attempt=1
[minimal] order=ORD-002 amount=$99.00 attempt=1
[minimal] order=ORD-003 amount=$10.00 attempt=1
[minimal] order=ORD-004 amount=$25.00 attempt=1
[minimal] order=ORD-005 amount=$77.77 attempt=1
[dlq] rejecting order=ORD-001 → DLQ
[dlq] dead-letter: order=ORD-001 reason=rejected deaths=1
[retry] order=ORD-001 attempt=1
[retry]   → transient failure, will Retry
[retry] order=ORD-001 attempt=2
[retry]   → success on retry
done

Source

//! Basic pub/sub examples covering all non-sequenced topology configurations (SQS backend).
//!
//! Demonstrates: `define_topic!`, all `Outcome` variants (`Ack`, `Retry`, `Reject`),
//! `publish`, `publish_with_headers`, `publish_batch`, and per-topic consumer
//! registration via `Broker<Sqs>::consumer_supervisor()`.
//!
//! Note: the earlier DLQ-consumer demo (`run_dlq::<DlqOrder>`) isn't wired
//! through the generic `ConsumerSupervisor<B>` wrapper yet — messages that
//! land in the DLQ here are simply left there for inspection.
//!
//! Spins up a LocalStack testcontainer automatically. Requires a running
//! Docker daemon and the `LOCALSTACK_AUTH_TOKEN` environment variable:
//!
//!     LOCALSTACK_AUTH_TOKEN=... cargo run --example sqs_basic_pubsub --features aws-sns-sqs
 
use std::collections::HashMap;
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::sns::SnsConfig;
use shove::{
    Broker, ConsumerOptions, DeadMessageMetadata, MessageHandler, MessageMetadata, Outcome, Sqs,
    TopologyBuilder, define_topic,
};
use testcontainers::ImageExt;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::localstack::LocalStack;
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderEvent {
    order_id: String,
    amount_cents: u64,
}
 
// ─── Topic definitions ──────────────────────────────────────────────────────
 
// 1. Minimal: just a queue. No DLQ, no hold queues.
//    Rejected messages are discarded with a warning log.
define_topic!(
    MinimalOrder,
    OrderEvent,
    TopologyBuilder::new("sqs-minimal-orders").build()
);
 
// 2. With DLQ: rejected messages land in a dead-letter queue for inspection.
define_topic!(
    DlqOrder,
    OrderEvent,
    TopologyBuilder::new("sqs-dlq-orders").dlq().build()
);
 
// 3. With hold queues + DLQ: escalating retry backoff (5 s → 30 s → 120 s).
//    Messages that exceed max_retries are sent to DLQ.
define_topic!(
    RetryOrder,
    OrderEvent,
    TopologyBuilder::new("sqs-retry-orders")
        .hold_queue(Duration::from_secs(5))
        .hold_queue(Duration::from_secs(30))
        .hold_queue(Duration::from_secs(120))
        .dlq()
        .build()
);
 
// ─── Handlers ───────────────────────────────────────────────────────────────
 
// Acks every message.
struct AckHandler;
 
impl MessageHandler<MinimalOrder> for AckHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[minimal] order={} amount=${:.2} attempt={}",
            msg.order_id,
            msg.amount_cents as f64 / 100.0,
            metadata.retry_count + 1,
        );
        Outcome::Ack
    }
}
 
// Rejects every message. Also implements handle_dead for DLQ processing.
struct RejectHandler;
 
impl MessageHandler<DlqOrder> for RejectHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, _metadata: MessageMetadata, _: &()) -> Outcome {
        println!("[dlq] rejecting order={} → DLQ", msg.order_id);
        Outcome::Reject
    }
 
    async fn handle_dead(&self, msg: OrderEvent, metadata: DeadMessageMetadata, _: &()) {
        println!(
            "[dlq] dead-letter: order={} reason={} deaths={}",
            msg.order_id,
            metadata.reason.as_deref().unwrap_or("unknown"),
            metadata.death_count,
        );
    }
}
 
// Retries once (simulates transient failure), then acks.
struct RetryHandler;
 
impl MessageHandler<RetryOrder> for RetryHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[retry] order={} attempt={}",
            msg.order_id,
            metadata.retry_count + 1,
        );
        if metadata.retry_count == 0 {
            println!("[retry]   → transient failure, will Retry");
            Outcome::Retry
        } else {
            println!("[retry]   → success on retry");
            Outcome::Ack
        }
    }
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let auth_token = match std::env::var("LOCALSTACK_AUTH_TOKEN") {
        Ok(t) => t,
        Err(_) => {
            eprintln!(
                "LOCALSTACK_AUTH_TOKEN is not set. This example requires a LocalStack Pro auth \
                 token:\n\n    export LOCALSTACK_AUTH_TOKEN=...\n"
            );
            std::process::exit(1);
        }
    };
 
    // Dummy AWS credentials for LocalStack.
    // SAFETY: called before any concurrent env access in this process.
    unsafe {
        std::env::set_var("AWS_ACCESS_KEY_ID", "test");
        std::env::set_var("AWS_SECRET_ACCESS_KEY", "test");
        std::env::set_var("AWS_REGION", "us-east-1");
    }
 
    let container = LocalStack::default()
        .with_env_var("LOCALSTACK_AUTH_TOKEN", auth_token)
        .start()
        .await?;
    let port = container.get_host_port_ipv4(4566).await?;
    let endpoint = format!("http://localhost:{port}");
 
    let broker = Broker::<Sqs>::new(SnsConfig {
        region: Some("us-east-1".into()),
        endpoint_url: Some(endpoint),
    })
    .await?;
 
    // ── Declare all topologies ──
    let topology = broker.topology();
    topology.declare::<MinimalOrder>().await?;
    topology.declare::<DlqOrder>().await?;
    topology.declare::<RetryOrder>().await?;
    println!("topologies declared\n");
 
    // ── Publish ──
    let publisher = broker.publisher().await?;
 
    // Single publish
    let order = OrderEvent {
        order_id: "ORD-001".into(),
        amount_cents: 5000,
    };
    publisher.publish::<MinimalOrder>(&order).await?;
    publisher.publish::<DlqOrder>(&order).await?;
    publisher.publish::<RetryOrder>(&order).await?;
 
    // Publish with custom headers
    let mut headers = HashMap::new();
    headers.insert("x-source".into(), "example".into());
    headers.insert("x-priority".into(), "high".into());
    let order2 = OrderEvent {
        order_id: "ORD-002".into(),
        amount_cents: 9900,
    };
    publisher
        .publish_with_headers::<MinimalOrder>(&order2, headers)
        .await?;
 
    // Batch publish
    let batch = vec![
        OrderEvent {
            order_id: "ORD-003".into(),
            amount_cents: 1000,
        },
        OrderEvent {
            order_id: "ORD-004".into(),
            amount_cents: 2500,
        },
        OrderEvent {
            order_id: "ORD-005".into(),
            amount_cents: 7777,
        },
    ];
    publisher.publish_batch::<MinimalOrder>(&batch).await?;
    println!("messages published\n");
 
    // ── Start consumers ──
    let mut supervisor = broker.consumer_supervisor();
    supervisor.register::<MinimalOrder, _>(AckHandler, ConsumerOptions::<Sqs>::new())?;
    supervisor.register::<DlqOrder, _>(RejectHandler, ConsumerOptions::<Sqs>::new())?;
    supervisor.register::<RetryOrder, _>(
        RetryHandler,
        ConsumerOptions::<Sqs>::new().with_max_retries(3),
    )?;
 
    // Let everything run, then shut down.
    let outcome = supervisor
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(10)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(5),
        )
        .await;
 
    println!("done");
    std::process::exit(outcome.exit_code());
}

Walkthrough

LocalStack setup

The example reads LOCALSTACK_AUTH_TOKEN from the environment and exits with a clear message if it is missing. Dummy AWS credentials (test/test) are set via std::env::set_var before any AWS SDK calls — LocalStack accepts these without validation. The container exposes the LocalStack endpoint on port 4566, which is fed into SnsConfig { endpoint_url: Some(endpoint) }.

`SnsConfig` and `Broker<Sqs>`

Broker::<Sqs>::new(SnsConfig { region, endpoint_url }) initialises the SNS/SQS client. Providing endpoint_url redirects all AWS API calls to LocalStack instead of the real AWS endpoints. In production you omit endpoint_url and rely on standard AWS credential/region resolution.

Three topology shapes

MinimalOrder has no DLQ — rejected messages are discarded. DlqOrder adds .dlq() so Outcome::Reject lands messages in a dead-letter queue; RejectHandler implements handle_dead to process those entries. RetryOrder chains .hold_queue(5s), .hold_queue(30s), and .hold_queue(120s) — on the SQS backend each stage maps to a separate SQS queue with a corresponding visibility timeout extension. with_max_retries(3) on the supervisor registration caps retry loops before DLQ routing.

Consumer supervisor on SQS

broker.consumer_supervisor() returns a ConsumerSupervisor\<Sqs\>. Each register call spawns a long-poll SQS consumer for that topic. Unlike AMQP, SQS consumers poll ReceiveMessage in a loop — there is no broker-push delivery. The supervisor handles the polling loop, visibility timeout management, and graceful drain on shutdown.

What to try next

Increase with_max_retries(3) to with_max_retries(0) — messages go directly to the DLQ on first failure.
Change Outcome::Retry to Outcome::Reject in RetryHandler — the message bypasses hold queues and goes straight to the DLQ.
Inspect the LocalStack management UI at http://localhost:4566/_localstack/health to see queues, message counts, and DLQ contents while the example runs.
See Guides: Retries for how hold queues and max-retries interact on SQS.