KinesisBroker

KinesisBroker connects a service to Kinesis Data Streams; new records configuration only, and the runtime builds the SDK client at startup.

KinesisBroker::new() does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: the AWS credentials and region, a local endpoint, and the lease store and owner id that let several instances share the shards. A setting the service leaves out comes from the AWS SDK's default chain.

At startup the runtime connects the broker and receives a ConnectedKinesisBroker. The connect loads the AWS configuration and builds the SDK client; it sends no request to Kinesis. Every subscription and every publisher of the service runs on that connected form, and the first requests go out when the subscriptions start. A handler that runs on a thread of its own publishes and checkpoints through a client built for that thread from the same configuration, so one busy runtime never holds up another's requests. At shutdown the runtime consumes the connected form, and the connection is closed.

#[subscriber("orders")] reads every shard of the stream orders. The descriptor KinesisStream carries the settings of a subscription, and the Subscriptions pages describe them. Records leave through the KinesisPublish policy, the broker's default.

Example

The service reads the stream orders. The region and the credentials come from the environment it runs in. cargo run --example kinesis_service --features testing -- run starts it.

Macros
Manual

examples/kinesis_service.rs
use ruststream_kinesis::prelude::*;

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

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(place);
    })
}

examples/manual/kinesis_service.rs
use ruststream_kinesis::prelude::*;

/// 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 app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(subscriber(KinesisStream::new("orders"), Place).build());
    })
}

Example​

Example