Startup

Startup opens every subscription: it resolves the queue URL, reads the queue's visibility timeout and writes a declared redrive policy.

Each subscription first checks its descriptor, before any call to AWS. An empty queue name, a wait above 20 seconds or a visibility outside one second to twelve hours fails startup with SqsError::InvalidQueue.

The subscription then resolves the queue name to its URL with GetQueueUrl, once, and the broker caches the answer. A queue URL in the descriptor is used as it is. A queue that does not exist fails startup with SqsError::Queue, naming the queue, unless the subscription creates it. A registration with a redrive policy writes the policy onto the queue next. A subscription that names no visibility reads the queue's own timeout with GetQueueAttributes, and holds every delivery under it.

Every request runs under the per-attempt timeout of 25 seconds that new sets on connect. A broker built from a config runs under the timeouts that config carries. The first request also fetches the credentials, so a missing credential or an unreachable endpoint fails startup too.

The in-process account a TestApp test runs on has every queue the service names, so a missing queue shows only against a running stack. A descriptor out of range fails TestApp::start as it fails a real startup.

Example

The subscription reads a queue the infrastructure created, and takes its visibility timeout from the queue.

Macros
Manual

examples/sqs_broker_startup.rs
#[subscriber(SqsQueue)]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order received");
    HandlerOutcome::ack()
}

/// Reads `queue`, which the infrastructure creates. The subscription names no visibility
/// timeout, so startup reads the queue's own.
fn service(broker: SqsBroker, queue: String) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(handle.name(queue).wait(Duration::from_secs(20)));
    })
}

examples/manual/sqs_broker_startup.rs
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(()))
    }
}

/// Reads `queue`, which the infrastructure creates. The subscription names no visibility
/// timeout, so startup reads the queue's own.
fn service(broker: SqsBroker, queue: String) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(Unnamed::<SqsQueue>::new(), Receive)
                .build()
                .name(queue)
                .wait(Duration::from_secs(20)),
        );
    })
}

Example​

Example