Prelude

One glob, ruststream_kinesis::prelude, brings the core prelude, the broker, the descriptor, positions, context keys and the publish surface.

use ruststream_kinesis::prelude::*; is the one import a Kinesis service writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds KinesisBroker, the descriptor KinesisStream with its mount-site steps KinesisSubscriberExt, and KinesisPosition with the core's Seeker and Positioned. It adds the contexts KinesisContext and KinesisBatchContext with their keys Position and SeekHandle, and the seeker type KinesisSeeker. With the dynamodb-lease feature it adds DynamoLeaseStore.

The publish policy comes under the name every broker crate uses: Publish is KinesisPublish. So a mount site reads the same on every broker. Two broker preludes in one file collide on that name; the prefixed type at the crate root tells them apart. The prelude also brings KinesisPublishSettings, KinesisPublishSteps and KinesisPublishOptions.

A handler body bounds its slots with capability traits, such as Out<impl Publisher, Slot>, and names no broker type. The one exception is a body that names a record's partition key: it needs KinesisPublishSteps in scope and the slot bounded Out<impl Publisher<Options = KinesisPublishOptions>, Slot>, both from this prelude.

Example

The descriptor, its read pause, the start position, the context key and the reply policy with its key all come from the one glob.

Macros
Manual

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

#[subscriber(
    KinesisStream::new("orders").poll_interval(Duration::from_millis(500)),
    start_at(KinesisPosition::horizon()),
    reply
)]
async fn confirm(order: &Order, Ctx(at): Ctx<Position>) -> Receipt {
    tracing::info!(order.id, position = ?at, "order confirmed");
    Receipt { order: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(confirm)
            .out_reply(Publish::default())
            .partition_key("receipts-v1");
    })
}

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

/// Confirms each order with a receipt, and logs where the order sits in the stream.
struct Confirm;

impl Handle<Order, Receipt, (), KinesisContext> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, KinesisContext>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(order.id, position = ?ctx.context(Position), "order confirmed");
        ready(Ok(Receipt { order: order.id }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        let orders = KinesisStream::new("orders").poll_interval(Duration::from_millis(500));
        b.include(
            subscriber(orders, Confirm)
                .reply()
                .build()
                .start_at(KinesisPosition::horizon()),
        )
        .out_reply(Publish::default())
        .partition_key("receipts-v1");
    })
}

Example​

Example