Getting started

This guide installs ruststream-sqs-sns, starts a local SQS and runs a first service against it.

Installation

Cargo.toml

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "logging"] }
ruststream-sqs-sns = "0.7"
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

[dev-dependencies]
ruststream-sqs-sns = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt"] }

ruststream is the framework: macros gives the attributes, json the default codec, and logging prints the service's log to the terminal. ruststream-sqs-sns is the SQS broker, built on the official aws-sdk-sqs client. Its testing feature adds the in-process SQS that tests run on.

The broker crate has these features, all off by default:

Feature	What it gives
sns	publishing to SNS topics with SnsPublish, over aws-sdk-sns (Fan-out with SNS)
asyncapi	the sqs and sns bindings in the generated AsyncAPI document
testing	the in-process SQS behind TestApp

A local SQS

LocalStack runs SQS and SNS in one container:

compose.yaml

services:
  localstack:
    image: localstack/localstack:4.14.0
    ports:
      - "4566:4566"
    environment:
      SERVICES: sqs,sns
      SQS_ENDPOINT_STRATEGY: path

docker compose up -d

The stack answers on http://localhost:4566 and starts with no queues.

The first service

Macros

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

// The local stack starts empty, so the descriptor creates the queue on subscribe.
#[subscriber(SqsQueue::new("orders").create_if_missing())]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(take_order);
        },
    )
}

Manual

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    amount_cents: u64,
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            // The local stack starts empty, so the descriptor creates the queue on subscribe.
            b.include(
                subscriber("orders", TakeOrder)
                    .map_source(|_| SqsQueue::new("orders").create_if_missing())
                    .build(),
            );
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

take_order reads JSON orders from the orders queue. It acknowledges an order with an amount and drops one without. SqsQueue names the queue, and create_if_missing creates it when the subscription opens. A production service leaves that step out and takes its queues from its infrastructure. SqsBroker::new only records the configuration. endpoint, region and test_credentials point it at the local stack. The SDK clients are built when the service starts. #[ruststream::app] turns app into main; in the manual form, run_main does the same.

Run the service:

cargo run -- run

Send an order from another terminal with the AWS CLI:

export AWS_ENDPOINT_URL=http://localhost:4566 AWS_REGION=us-east-1 \
  AWS_ACCESS_KEY_ID=test AWS_SECRET_ACCESS_KEY=test
aws sqs send-message \
  --queue-url "$(aws sqs get-queue-url --queue-name orders --output text)" \
  --message-body '{"id":1,"amount_cents":1250}'

The service logs order taken with the order's id. Ctrl+C stops it.

The first test

The example file also holds two tests. Each starts the same app with TestApp::start, publishes an order and checks how take_order settled it. SqsBroker connects to an in-process SQS there, so the tests need no stack:

cargo test

Testing an SQS service walks through these tests and what the in-process SQS models.

Coming from aws-sdk-sqs

A service on aws-sdk-sqs writes its own receive loop. Here is where each part goes in RustStream:

aws-sdk-sqs	RustStream
aws_config::load_from_env and Client::new	SqsBroker::new, or SqsBroker::from_config with an SdkConfig of your own (Running in production)
get_queue_url	the queue name in SqsQueue::new, resolved once at startup
a receive_message().wait_time_seconds(20) loop	a #[subscriber] handler (Consuming queues)
deserializing message.body()	the handler's parameter type, decoded by the codec
delete_message after the work	the handler's outcome, HandlerOutcome::ack (Acknowledgement and delivery guarantees)
change_message_visibility	HandlerOutcome::retry_after (Retries and dead-letter queues)
a RedrivePolicy set in the console	max_attempts and dead_letter at the mount site
send_message	a reply or an Out parameter, sent through SqsPublish
message_group_id and message_deduplication_id	the publish steps group_id and deduplication_id (FIFO queues)