KinesisStream

KinesisStream describes one subscription by stream name, with its read pause and the shards it provisions when the stream is missing.

KinesisStream::new takes the stream name, and both settings are optional. poll_interval is the pause between reads of an idle shard, one second by default. create_if_missing creates the stream with the shards it names when the service lacks it; by default a missing stream stops the service at startup. The descriptor goes where a stream name goes: the first argument of #[subscriber], or the first argument of subscriber(source, body) in the manual form. The same two settings are also mount-site steps. The subscription's name is the stream: a test reads its deliveries under that name, and the AsyncAPI document reports that channel with both settings. A retry copy goes back to the same stream with no setting at the mount site, because a stream is both what a subscription reads and what a publish writes to.

Example

Every payment of payments is captured; the stream is read every half second when it is quiet.

Macros
Manual

examples/kinesis_subscriptions_stream.rs
/// Captures each payment. The descriptor names the stream and the settings of its reads.
#[subscriber(
    KinesisStream::new("payments")
        .poll_interval(Duration::from_millis(500))
        .create_if_missing(4)
)]
async fn capture(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.id, "payment captured");
    HandlerOutcome::ack()
}

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

examples/manual/kinesis_subscriptions_stream.rs
/// Captures each payment. The descriptor names the stream and the settings of its reads.
struct Capture;

impl Handle<Payment> for Capture {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payment.id, "payment captured");
        ready(Ok(()))
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        let payments = KinesisStream::new("payments")
            .poll_interval(Duration::from_millis(500))
            .create_if_missing(4);
        b.include(subscriber(payments, Capture).build());
    })
}

Example​

Example