SqsQueue

SqsQueue describes one subscription by queue name or URL, with its long poll, visibility timeout and queue creation.

SqsQueue::new takes a queue name or a queue URL (https://sqs...). A name resolves through GetQueueUrl once, when the subscription opens; a URL is used as it is. Three steps set the rest, and each has a default: wait (20 seconds), visibility (the queue's own timeout) and create_if_missing (off). The descriptor goes where the name goes, as the first argument of #[subscriber(..)]. The subscription's name is what the descriptor names: a test reads its deliveries under that name. A descriptor that cannot form a subscription stops the service at startup with SqsError::InvalidQueue, before any call to AWS: an empty name, a wait over 20 seconds, or a visibility outside one second to twelve hours. The same three steps are on the mount chain through SqsSubscription. The manual form builds the descriptor and hands it to the chain with map_source.

Example

Orders read with a ten second long poll, each hidden from other consumers for a minute.

Macros
Manual

examples/sqs_subscriptions_queue.rs
/// Reads the queue `orders`: each receive waits up to ten seconds for a message, and a received
/// order stays hidden from other consumers for a minute.
#[subscriber(
    SqsQueue::new("orders")
        .wait(Duration::from_secs(10))
        .visibility(Duration::from_secs(60))
)]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

fn service(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
    })
}

examples/manual/sqs_subscriptions_queue.rs
/// Reads the queue `orders`: each receive waits up to ten seconds for a message, and a received
/// order stays hidden from other consumers for a minute.
struct Place;

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

fn service(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let orders = SqsQueue::new("orders")
            .wait(Duration::from_secs(10))
            .visibility(Duration::from_secs(60));
        b.include(subscriber("orders", Place).build().map_source(|_| orders));
    })
}

Example​

Example