Skip to main content

Consuming queues

This guide reads SQS queues: long polling, the visibility timeout of a slow handler, parallel handlers, batches, and one handler on several queues.

The example is a warehouse service. It packs orders, books payments and records changes to orders.

A queue and its long poll​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    items: u32,
}

/// Packs an order. Packing takes up to a minute, so each receive asks for a minute of
/// visibility, and four orders are packed at once.
#[subscriber(
    SqsQueue::new("orders")
        .visibility(Duration::from_secs(60))
        .create_if_missing(),
    workers(4)
)]
async fn pack(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.items, "order packed");
    HandlerOutcome::ack()
}

A subscription long-polls its queue with ReceiveMessage. Each call waits up to 20 seconds for a message, the most SQS allows. wait sets a shorter wait. A queue is named by its name or by its URL, and a name is resolved with GetQueueUrl once, at startup.

A single-message subscription still asks for up to ten messages per call, because SQS bills per request. It hands them to the handler one at a time. A background task receives the next batch while the handler works, and it never runs more than one batch ahead.

Visibility and a slow handler​

A received message stays invisible to other consumers for the queue's visibility timeout. If it is not deleted by then, SQS hands it out again. visibility asks for a timeout of its own on each receive, between one second and twelve hours. SQS counts it in whole seconds, so a fraction rounds up to the next second. pack asks for a minute. Without the step, the subscription reads the queue's own timeout once, when it opens.

A handler may run longer than the timeout. While a delivery is alive, the crate extends its visibility halfway through each timeout, so a slow handler does not see its message handed to another consumer. SQS keeps a message hidden for at most twelve hours from its receive. A handler that runs that long gets one last extension to the end of them and a warning in the log, and the message then goes to another consumer.

Parallel handlers​

workers(4) runs up to four deliveries of orders at once. The four handlers share the messages each receive returns. A standard queue has no order to keep, so nothing else is needed. FIFO queues keep one message group in order with workers(n, by_key).

Batches​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    order: u64,
    amount_cents: u64,
}

/// Books the payments one receive returned, in one write to the ledger.
#[subscriber(SqsQueue::new("payments"))]
async fn reconcile(payments: &[Payment]) -> HandlerOutcome {
    let total: u64 = payments.iter().map(|payment| payment.amount_cents).sum();
    tracing::info!(count = payments.len(), total, "payments booked");
    HandlerOutcome::ack()
}

A handler that takes a slice is a batch handler. The size its mount site names becomes MaxNumberOfMessages, so one receive is one batch. A batch holds what the queue had at the time, and a long poll that returns nothing yields no batch. ReceiveMessage returns at most ten messages, so a larger size is lowered to ten and the service logs it once. Each message of a batch is settled on its own; Acknowledgement shows the outcomes.

One handler, several queues​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Change {
    order: u64,
    by: String,
}

/// Records a change to an order. The definition fixes the kind of queue and leaves its name
/// to the mount site.
#[subscriber(SqsQueue)]
async fn audit(change: &Change) -> HandlerOutcome {
    tracing::info!(change.order, change.by, "change recorded");
    HandlerOutcome::ack()
}

#[subscriber(SqsQueue)] fixes the kind of source and leaves the queue's name open. The mount site names it with .name(..), once per queue.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(pack);
            // Up to ten payments per call: one receive is one batch.
            b.include(
                reconcile
                    .batch(nonzero!(10))
                    .wait(Duration::from_secs(20))
                    .create_if_missing(),
            );
            // One definition, two queues.
            b.include(audit.name("audit-eu").create_if_missing());
            b.include(audit.name("audit-us").create_if_missing());
        },
    )
}

The queue steps also sit at the mount site, after the framework's own steps, through SqsSubscription. An empty queue name, a wait over 20 seconds or a visibility outside its range stops the service at startup with SqsError::InvalidQueue, before any call to AWS. A name may hold characters SQS does not take: each character outside A-Z, a-z, 0-9, _ and - becomes -, and a .fifo suffix is kept.

create_if_missing creates a queue that does not exist. A production service usually takes its queues from its infrastructure and leaves the step out; a missing queue then stops it at startup.

What a delivery carries​

Each message attribute arrives as a header, and each header of a publish leaves as an attribute. Two headers are the crate's own. RECEIVE_COUNT_HEADER carries the queue's receive count of the message. PARTITION_KEY_HEADER carries the message group of a FIFO queue. An SQS body is text, so a binary payload travels base64-encoded with a marker attribute and arrives decoded. Base64 makes such a payload a third larger, and SQS refuses a message over its size limit.

The tests in the example file publish orders, payments and changes to the app in process. Testing an SQS service covers the harness.