KinesisSubscriberExt

KinesisSubscriberExt puts poll_interval and create_if_missing on the mount chain, after batch(n) and before start_at(..).

The two steps are the mount-site spellings of poll_interval and create_if_missing on KinesisStream. They suit a handler whose attribute names the stream and leaves the read settings to the code that mounts it. They change the subscription's descriptor, so they need the descriptor to still be there. The framework's own steps, such as workers(n) and batch(n), come first. start_at(..) comes last: it wraps the descriptor in a start position, and the Kinesis steps are not offered after it. A handler that names its start position in the attribute names the read settings on the descriptor instead. In the manual form the steps follow .build() on subscriber(source, body).

Example

The archive reads the whole retained stream with four workers, a 250 ms pause and a stream it creates on a bare stand.

Macros
Manual

examples/kinesis_subscriptions_ext.rs
/// Archives every audit event the stream still retains.
#[subscriber(KinesisStream::new("audit"))]
async fn archive(event: &AuditEvent) -> HandlerOutcome {
    tracing::info!(event.id, "archived");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // The framework's step first, this crate's settings next, the start position last.
        b.include(
            archive
                .workers(nonzero!(4))
                .poll_interval(Duration::from_millis(250))
                .create_if_missing(1)
                .start_at(KinesisPosition::horizon()),
        );
    })
}

examples/manual/kinesis_subscriptions_ext.rs
/// Archives every audit event the stream still retains.
struct Archive;

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

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // The framework's step first, this crate's settings next, the start position last.
        b.include(
            subscriber(KinesisStream::new("audit"), Archive)
                .workers(nonzero!(4))
                .build()
                .poll_interval(Duration::from_millis(250))
                .create_if_missing(1)
                .start_at(KinesisPosition::horizon()),
        );
    })
}

Example​

Example