Subscriptions

A queue name in #[subscriber] long-polls that queue; SqsQueue carries every other setting of the subscription.

#[subscriber("orders")] reads the queue orders. The name resolves to the queue's URL through GetQueueUrl once, when the subscription opens. A bare name takes the descriptor's defaults: a 20 second long poll and the queue's own visibility timeout. #[subscriber(SqsQueue)] fixes the kind and leaves the queue to the mount site: b.include(audit.name("audit-eu")) reads audit-eu, so one definition serves several queues. SQS takes only [A-Za-z0-9_-] in a queue name, so every other character of the name becomes - on the way, and a .fifo suffix survives: order.events reads the queue order-events. Two subscriptions on one queue compete for its messages, and each message goes to one of them. A subscription receives at most one batch ahead of its handler. A queue that does not exist stops the service at startup, unless the subscription creates it. Once the service runs, a queue deleted under a subscription ends that subscription, and any other receive error is retried after one second. The manual form passes the name to subscriber(name, body), and Unnamed<SqsQueue> in place of the name leaves it to name at the mount site.

Example

The queue orders, read with the defaults.

Macros

examples/sqs_subscriptions_name.rs

/// Reads the queue `orders` with the descriptor's defaults: a 20 second long poll and the
/// queue's own visibility timeout.
#[subscriber("orders")]
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);
    })
}

Manual

examples/manual/sqs_subscriptions_name.rs

/// Reads the queue `orders` with the descriptor's defaults: a 20 second long poll and the
/// queue's own visibility timeout.
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| {
        b.include(subscriber("orders", Place).build());
    })
}

One definition, two queues

The audit handler reads audit-eu and audit-us, each a subscription of its own.

Macros

examples/sqs_subscriptions_name.rs

/// The definition fixes the kind of subscription and leaves the queue to the mount site.
#[subscriber(SqsQueue)]
async fn audit(entry: &Entry) -> HandlerOutcome {
    tracing::info!(entry.id, "entry audited");
    HandlerOutcome::ack()
}

fn audits(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        b.include(audit.name("audit-eu"));
        b.include(audit.name("audit-us"));
    })
}

Manual

examples/manual/sqs_subscriptions_name.rs

/// The definition fixes the kind of subscription and leaves the queue to the mount site.
struct Audit;

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

fn audits(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(Unnamed::<SqsQueue>::new(), Audit)
                .build()
                .name("audit-eu"),
        );
        b.include(
            subscriber(Unnamed::<SqsQueue>::new(), Audit)
                .build()
                .name("audit-us"),
        );
    })
}