Enum KinesisPosition 

pub enum KinesisPosition {
    Horizon,
    Latest,
    Timestamp(u64),
    Sequence {
        shard: String,
        sequence: String,
    },
}

A position in the stream’s retained log: the whole start vocabulary of this broker, accepted by Seeker::seek and by the start_at(..) clause of #[subscriber(..)].

Repositioning resets the checkpoint bookkeeping of every shard it moves: acknowledgements of records delivered before the seek stop advancing the watermark, so a stale checkpoint cannot drag the cursor back over the position just taken. Records from the new position onward are delivered again, which at-least-once permits.

Without a position a subscription resumes from the stored checkpoint of each shard, and starts at the tip on a shard that has none. A position given to start_at(..) applies to the shards the subscription opens with; a shard it takes on later resumes from its checkpoint.

Opening a subscription returns once every shard it took has its reader and its starting cursor, so the tip is the moment the subscription opened: a record published after that is read.

Examples

use ruststream_kinesis::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

/// Rebuilds a read model from every record the stream still retains, on every shard.
#[subscriber(KinesisStream::new("orders"), start_at(KinesisPosition::horizon()))]
async fn backfill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "backfilled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(backfill);
    })
}

Variants

Horizon

The trim horizon: everything the stream still retains.

Stream-wide - it applies to every shard the subscription reads when it lands, and to a shard whose reader starts later without a checkpoint.

Latest

The tip: only records published after the reposition.

Stream-wide, and the position a shard without a checkpoint starts at by default. A child of a split or a merge whose parent was read to its end opens at its own beginning instead, so nothing written to it before its reader started is skipped.

Timestamp(u64)

The first record at or after this timestamp, in milliseconds since the Unix epoch.

Stream-wide; each shard opens at its own first record from that instant. The instant is matched against the arrival timestamp the service stamped on the record, and the service calls that stamp approximate: records that arrived around the same moment are not separated exactly.

Sequence

Exactly one record on one shard.

This is the pinned form the framework defines for captured positions (Positioned::position): seeking to one redelivers that very record. It addresses a single shard, so it moves that shard’s reader only, the way a partitioned log seeks per partition, and the shard must be owned and live.

Fields

shard: String

The shard the position lives on.

sequence: String

The record’s sequence number.

Implementations

impl KinesisPosition

pub const fn horizon() -> Self

The trim horizon, for every shard: see KinesisPosition::Horizon.

pub const fn latest() -> Self

The tip, for every shard: see KinesisPosition::Latest.

pub const fn timestamp(millis: u64) -> Self

A wall-clock instant (milliseconds since the Unix epoch), for every shard: see KinesisPosition::Timestamp.

pub fn sequence(shard: impl Into<String>, sequence: impl Into<String>) -> Self

One record on one shard: see KinesisPosition::Sequence.

Captured positions come from Positioned::position; this constructor is for a sequence number carried in from elsewhere (an operator’s replay request, say).

Trait Implementations

impl Clone for KinesisPosition

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for KinesisPosition

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for KinesisPosition

impl PartialEq for KinesisPosition

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for KinesisPosition