Channels

A subscription's channel reports its stream, read pause and provisioned shards; a publishing position's channel reports the stream it writes.

A KinesisStream writes three fields into its channel's x-ruststream-kinesis object. stream is the stream it reads, and pollIntervalMs the pause between reads of an idle shard, in milliseconds. shardCount is the number of shards create_if_missing gives the stream, and it is there only when the descriptor creates the stream.

A channel the service publishes to reports stream, the stream its records land in. The policy holds no destination, so the stream is the one the mount site resolved: a reply type's #[outgoing(name)], a reply("stream") clause, or the name of a slot entry.

Example

The subscription reads orders every quarter of a second and creates it with two shards, and its replies land in receipts.

Macros

examples/kinesis_asyncapi.rs

/// Reads `orders` every quarter of a second, and creates it with two shards where it is missing.
#[subscriber(
    KinesisStream::new("orders")
        .poll_interval(Duration::from_millis(250))
        .create_if_missing(2),
    reply
)]
async fn confirm(order: &Order) -> Receipt {
    Receipt { id: order.id }
}

Manual

examples/manual/kinesis_asyncapi.rs

/// Confirms each order with a receipt.
struct Confirm;

impl Handle<Order, Receipt> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        ready(Ok(Receipt { id: order.id }))
    }
}

/// Reads `orders` every quarter of a second, and creates it with two shards where it is missing.
fn orders() -> KinesisStream {
    KinesisStream::new("orders")
        .poll_interval(Duration::from_millis(250))
        .create_if_missing(2)
}

The channel orders in the document:

tests/snippets/asyncapi-channel.json

{
  "x-ruststream-kinesis": {
    "pollIntervalMs": 250,
    "shardCount": 2,
    "stream": "orders"
  }
}

The channel receipts:

tests/snippets/asyncapi-publish-channel.json

{
  "x-ruststream-kinesis": {
    "stream": "receipts"
  }
}