Subscriptions

A stream name in #[subscriber] reads every shard of that stream; KinesisStream carries every other setting of the subscription.

#[subscriber("orders")] reads the stream orders with the descriptor's defaults: KinesisStream::new("orders"). A stream is named by its name, which the service limits to 128 characters of letters, digits, _, . and -. The subscription finds the stream's shards, takes a lease on each one and reads each leased shard with its own reader. A shard resumes from its stored checkpoint, and a shard without one opens at the tip; start_at names another position. The subscription is open once every shard it took has a reader and a starting cursor, so a record published after startup reaches it. A stream the service does not have stops the service at startup, unless the descriptor creates it. Delivery is at-least-once, and an acknowledgement moves the shard's checkpoint, as Checkpoint describes. The manual form passes the name to subscriber(name, body).

Example

The stream orders, read by one handler.

Macros

examples/kinesis_subscriptions_name.rs

/// Reads every shard of the stream `orders`.
#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

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

Manual

examples/manual/kinesis_subscriptions_name.rs

/// Reads every shard of the stream `orders`.
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: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Place).build());
    })
}