Batches

A slice handler gets up to batch(n) records; n becomes the GetRecords limit, and a short batch goes out 50 ms after its first record.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. Every shard reader asks GetRecords for at most n records, so one read never fetches more than one batch holds; the service caps a read at 10,000 records. A batch that is still filling goes to the handler 50 ms after its first record, so on a quiet stream a batch is often smaller than n. One batch mixes the records of every shard this instance reads. One outcome settles every record of the batch. A Vec<HandlerOutcome> settles record i with outcome i, and each record moves the checkpoint of its own shard. A retry_after in a batch becomes a copy published back to the stream after the delay. A small batch means small reads against the same five reads a second per shard, which poll_interval paces. A batch body reads a KinesisBatchContext, which carries the seek handle and no position of a single record. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles record by record.

Example

Orders are stored up to 500 at a time; refunds are reviewed up to 100 at a time.

Macros

examples/kinesis_batches.rs

/// Stores a batch of orders in one round-trip and settles the whole batch with one outcome.
#[subscriber(KinesisStream::new("orders"))]
async fn digest(batch: &[Order]) -> HandlerOutcome {
    tracing::info!(count = batch.len(), "orders stored");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // The framework's step first, this crate's read setting after it.
        b.include(
            digest
                .batch(nonzero!(500))
                .poll_interval(Duration::from_millis(500)),
        );
        b.include(review.batch(nonzero!(100)));
    })
}

Manual

examples/manual/kinesis_batches.rs

/// Stores a batch of orders in one round-trip and settles the whole batch with one outcome.
struct Digest;

impl Handle<[Order]> for Digest {
    fn handle(
        &self,
        batch: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = batch.len(), "orders stored");
        ready(Ok(()))
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // The framework's step first, this crate's read setting after it.
        b.include(
            subscriber(KinesisStream::new("orders"), Digest)
                .batch(nonzero!(500))
                .build()
                .poll_interval(Duration::from_millis(500)),
        );
        b.include(
            subscriber(KinesisStream::new("refunds"), Refunds)
                .batch(nonzero!(100))
                .build(),
        );
    })
}

One outcome per record

Only the unapproved refund comes back, as a copy published after the delay.

Macros

examples/kinesis_batches.rs

/// One outcome per refund, in slice order: only the unapproved ones come back.
#[subscriber(KinesisStream::new("refunds"))]
async fn review(batch: &[Refund]) -> Vec<HandlerOutcome> {
    batch
        .iter()
        .map(|refund| {
            if refund.approved {
                HandlerOutcome::ack()
            } else {
                HandlerOutcome::retry_after(RECHECK)
            }
        })
        .collect()
}

Manual

examples/manual/kinesis_batches.rs

/// One outcome per refund, in slice order: only the unapproved ones come back.
struct Refunds;

impl Handle<[Refund]> for Refunds {
    fn handle(
        &self,
        batch: &[Refund],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        ready(Err(batch
            .iter()
            .map(|refund| {
                if refund.approved {
                    HandlerOutcome::ack()
                } else {
                    HandlerOutcome::retry_after(RECHECK)
                }
            })
            .collect()))
    }
}