EntryId

EntryId is a stream entry's <milliseconds>-<sequence> id; the EntryId key binds the id a delivery was read at.

The server assigns the id at XADD. Its first half, milliseconds, is the server clock at that moment. The second, sequence, tells apart the entries added within one millisecond. Ids compare the way the stream orders its entries. EntryId::new(milliseconds, sequence) builds one, and parsing a string checks it, so a value is always a well-formed id. The milliseconds half alone parses too: "1700000000000" is 1700000000000-0. to_string writes the form Redis takes. previous is the id right below, and EntryId::ZERO is below every entry. Ctx<keys::EntryId> binds the id of the delivery, and RedisGroupPosition::after(id) moves a group past it. A batch spans many entries, so a batch context carries no entry id.

Example

Each receipt names the stream entry it billed and when the server added it.

Macros
Manual

examples/redis_positions_entry_id.rs
/// The receipt names the entry it bills, so a duplicate bill is traced to one stream entry.
#[subscriber(RedisStream::new("orders").group("billing"), reply)]
async fn bill(order: &Order, Ctx(entry): Ctx<keys::EntryId>) -> Receipt {
    tracing::info!(order.id, %entry, "order billed");
    Receipt {
        order: order.id,
        entry: entry.to_string(),
        // The first half of the id is the server clock at `XADD`.
        added_ms: entry.milliseconds(),
    }
}

examples/manual/redis_positions_entry_id.rs
/// The receipt names the entry it bills, so a duplicate bill is traced to one stream entry.
struct Bill;

impl Handle<Order, Receipt, (), StreamContext> for Bill {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, StreamContext>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        let entry = ctx.context(keys::EntryId);
        tracing::info!(order.id, %entry, "order billed");
        ready(Ok(Receipt {
            order: order.id,
            entry: entry.to_string(),
            // The first half of the id is the server clock at `XADD`.
            added_ms: entry.milliseconds(),
        }))
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("billing");
        b.include(subscriber(orders, Bill).reply().build());
    })
}

Example​

Example