Kinesis bindings

With the asyncapi feature the generated document speaks Kinesis: the server's host, and an x-ruststream-kinesis object on each channel and keyed message.

The core builds the AsyncAPI document from the app, before anything connects. The specification defines no Kinesis binding, and its list of binding keys is closed. So this crate writes its own objects under the extension key x-ruststream-kinesis. A server reports the host clients connect to. A channel reports the stream behind it, and a subscription's channel its read settings too. A message reports the partition key the mount site fixed.

Everything comes from the service's own declarations: the broker, the descriptors and the policies. The document is built without a connection, so an existing stream's real shard count is not in it. The document is published and shared, so it carries no credential. The binary #[ruststream::app] writes prints the document with asyncapi gen.

Example

The dependency line:

ruststream-kinesis = { version = "0.7", features = ["asyncapi"] }

The service registers its broker under the label kinesis, the key of its server in the document. cargo run --example kinesis_asyncapi --features asyncapi -- asyncapi gen prints the document.

Macros

examples/kinesis_asyncapi.rs

fn service(broker: KinesisBroker) -> RustStream {
    // The label is the server's key in the document.
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker_labeled("kinesis", broker, |b| {
        // Every receipt carries one key, so they share a shard and keep their order.
        b.include(confirm)
            .out_reply(Publish::default())
            .partition_key("tenant-acme");
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(broker())
}

Manual

examples/manual/kinesis_asyncapi.rs

fn service(broker: KinesisBroker) -> RustStream {
    // The label is the server's key in the document.
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker_labeled("kinesis", broker, |b| {
        // Every receipt carries one key, so they share a shard and keep their order.
        b.include(subscriber(orders(), Confirm).reply().build())
            .out_reply(Publish::default())
            .partition_key("tenant-acme");
    })
}

fn app() -> impl App {
    service(broker())
}