Headers

Every delivery carries its sequence number, its shard and its partition key as headers, where the framework's per-key dispatch finds the key.

Header	Constant	Value
`kinesis-sequence-number`	`SEQUENCE_HEADER`	the record's sequence number on its shard, a decimal string too wide for an integer
`kinesis-shard-id`	`SHARD_HEADER`	the shard the record was read from, such as `shardId-000000000000`
`partition-key`	`PARTITION_KEY_HEADER`	the record's partition key

The subscription writes the three headers from the record's own fields, next to the headers the producer wrote in the envelope. A handler reads them from ctx.headers(), as text with get_str. The sequence number and the shard name the same record that the Position key names.

The partition key under partition-key is the key the core's IncomingMessage::partition_key reports. So workers(n, by_key) runs the records of one key one after another on one lane, in the order the shard holds them, while records of other keys run beside them. A retry copy carries the delivery's headers, so it keeps the record's partition key.

Example

The meter is each reading's partition key, so the readings of one meter are billed in order on four lanes.

Macros
Manual

examples/kinesis_context_headers.rs
use ruststream_kinesis::{PARTITION_KEY_HEADER, SEQUENCE_HEADER, SHARD_HEADER};

/// Bills each meter reading. The meter is the record's partition key, so the readings of one
/// meter run one after another on four lanes, and in the order the stream holds them.
#[subscriber(KinesisStream::new("meters"), workers(4, by_key))]
async fn bill(
    reading: &Reading,
    ctx: &mut Context<'_>,
    Out(bills): Out<impl Publisher, Bills>,
) -> HandlerOutcome {
    let headers = ctx.headers();
    let (Some(meter), Some(shard), Some(sequence)) = (
        headers.get_str(PARTITION_KEY_HEADER),
        headers.get_str(SHARD_HEADER),
        headers.get_str(SEQUENCE_HEADER),
    ) else {
        return HandlerOutcome::drop();
    };
    tracing::info!(meter, shard, sequence, reading.kwh, "reading billed");
    let charge = Bill {
        meter: meter.to_owned(),
        kwh: reading.kwh,
    };
    if bills.message(&charge).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kinesis_context_headers.rs
use ruststream_kinesis::{PARTITION_KEY_HEADER, SEQUENCE_HEADER, SHARD_HEADER};

/// Bills each meter reading. The meter is the record's partition key, so the readings of one
/// meter run one after another on four lanes, and in the order the stream holds them.
struct Charge;

impl<Entry> Handle<Reading, (), Outs<(Entry,)>> for Charge
where
    Entry: OutEntry<Bills, Wire: Publisher>,
{
    async fn handle(
        &self,
        reading: &Reading,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let headers = ctx.headers();
        let (Some(meter), Some(shard), Some(sequence)) = (
            headers.get_str(PARTITION_KEY_HEADER),
            headers.get_str(SHARD_HEADER),
            headers.get_str(SEQUENCE_HEADER),
        ) else {
            return Err(HandlerOutcome::drop());
        };
        tracing::info!(meter, shard, sequence, reading.kwh, "reading billed");
        let bill = Bill {
            meter: meter.to_owned(),
            kwh: reading.kwh,
        };
        outs.get(Bills)
            .message(&bill)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(
            subscriber(KinesisStream::new("meters"), Charge)
                .workers_by_key(nonzero!(4))
                .build(),
        )
        .out(Bills, Publish::default())
        .build();
    })
}

Example​

Example