SqsBroker

SqsBroker connects a service to Amazon SQS; new only records the configuration, and the runtime builds the SDK clients at startup.

SqsBroker::new takes no arguments and does no I/O, so the whole service fits the synchronous #[app] builder. Left as it is, the broker takes the region, the credentials and the endpoint from the AWS SDK's default chain: the environment, the shared profile files, and the instance or container role. The credentials page shows a configuration the service builds itself, and the endpoint page a local stack.

At startup the runtime connects the broker and receives a ConnectedSqsBroker. The connect loads the AWS configuration and builds the SQS client, and the SNS client with the sns feature. Every subscription and every publisher of the service opens against that connected form, and startup resolves each subscription's queue before the service reads anything. At shutdown the runtime marks the connection closed.

A request goes out through the SDK clients of the runtime that sends it. A handler on a dedicated thread, set with threads(n), gets clients of its own on its first request, built from the same configuration and sharing its credentials. Its connections stay in its own pool, so the broker's requests never wait on a thread that is busy computing.

#[subscriber("orders")] reads the queue orders with the default settings. SqsQueue carries the settings of one subscription, such as the long-polling wait and the visibility timeout.

Example

The handler reads the queue orders, each receive waiting up to 20 seconds for a message.

Macros
Manual

examples/sqs_service.rs
/// Reads the queue `orders`, each receive waiting up to 20 seconds for a message.
///
/// A fresh local stack has no queues, so the descriptor creates the queue at startup. A
/// production service drops that step and takes the queue from its infrastructure.
#[subscriber(
    SqsQueue::new("orders")
        .wait(Duration::from_secs(20))
        .create_if_missing()
)]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order received");
    HandlerOutcome::ack()
}

examples/manual/sqs_service.rs
/// Logs each order it receives.
struct Receive;

impl Handle<Order> for Receive {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order received");
        ready(Ok(()))
    }
}

The app mounts it on a broker that reads its configuration from the environment. With AWS_ENDPOINT_URL=http://localhost:4566, AWS_REGION and test keys set, cargo run --example sqs_service --features testing -- run starts it against a local stack.

Macros
Manual

examples/sqs_service.rs
#[app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(SqsBroker::new(), |b| {
        b.include(handle);
    })
}

examples/manual/sqs_service.rs
/// Reads the queue `orders`, each receive waiting up to 20 seconds for a message. A fresh local
/// stack has no queues, so the subscription creates the queue at startup.
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(SqsBroker::new(), |b| {
        b.include(
            subscriber(Unnamed::<SqsQueue>::new(), Receive)
                .build()
                .name("orders")
                .wait(Duration::from_secs(20))
                .create_if_missing(),
        );
    })
}

Example​

Example