KinesisPosition

KinesisPosition names a place in the stream: the trim horizon, the tip, an arrival timestamp on every shard, or one record of one shard.

A stream keeps its records for its retention period, so a subscription can be moved through them, at startup with start_at or from a handler with a seek. By default each shard resumes from its checkpoint, and a shard without one opens at the tip.

Position	Scope	Meaning
`KinesisPosition::horizon()`	every shard	the trim horizon: the oldest record the stream still retains
`KinesisPosition::latest()`	every shard	the tip: only records published from now on
`KinesisPosition::timestamp(millis)`	every shard	on each shard, the first record that arrived at or after that instant, in milliseconds since the Unix epoch
`KinesisPosition::sequence(shard, sequence)`	one shard	exactly one record, by its shard id and sequence number

The first three apply to every shard the subscription reads. A timestamp is matched against the arrival time the service stamped on each record, which the service calls approximate: records that arrived within the same moment are not told apart. A sequence position names one shard and moves no other. A delivery's own place is a sequence position: Ctx<Position> binds it, and a seek there reads that record again. The type is also what the core's Positioned returns for a delivered record.

Example

A report is rebuilt from the readings that arrived since an instant its configuration names.

Macros
Manual

examples/kinesis_positions_timestamp.rs
/// Adds a reading to the report.
#[subscriber(KinesisStream::new("readings"))]
async fn report(reading: &Reading) -> HandlerOutcome {
    tracing::info!(reading.sensor, reading.value, "reading reported");
    HandlerOutcome::ack()
}

/// The report opens every shard at the first reading that arrived at or after `since`, in
/// milliseconds since the Unix epoch.
fn service(broker: KinesisBroker, since: u64) -> RustStream {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(report.start_at(KinesisPosition::timestamp(since)));
    })
}

examples/manual/kinesis_positions_timestamp.rs
/// Adds a reading to the report.
struct Report;

impl Handle<Reading> for Report {
    fn handle(
        &self,
        reading: &Reading,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(reading.sensor, reading.value, "reading reported");
        ready(Ok(()))
    }
}

/// The report opens every shard at the first reading that arrived at or after `since`, in
/// milliseconds since the Unix epoch.
fn service(broker: KinesisBroker, since: u64) -> RustStream {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(KinesisStream::new("readings"), Report)
                .build()
                .start_at(KinesisPosition::timestamp(since)),
        );
    })
}

Example​

Example