Struct Stream 

pub struct Stream { /* private fields */ }

A log of entries in ID order.

Implementations

impl Stream

pub fn new() -> Stream

An empty stream.

pub fn freeze(&self, out: &mut Vec<u8>)

Write the stream out as the bytes a tier can hold, for crate::keyspace::Keyspace to hand back to Stream::thaw.

The nodes go out as the listpacks they already are, one master ID and one blob each. That is the whole point of the node layout: a run of entries is already a flat sequence of bytes with no pointers in it, so freezing one is a copy and thawing it is a length check. Only the counters and the consumer groups need a form of their own.

pub fn thaw(bytes: &[u8]) -> Result<Stream, Broken>

Read back a stream Stream::freeze wrote.

What is inside a node is checked as far as Listpack::from_bytes checks it, which is that the header, the lengths and the terminator all agree, and no further. Every walk over a node’s contents already returns early on anything it does not understand rather than trusting what it finds, so a structurally sound listpack full of nonsense answers an empty range instead of panicking. That is the same trust a node written by Redis and loaded from an RDB file gets today.

pub fn len(&self) -> u64

How many live entries there are, which is XLEN.

pub fn is_empty(&self) -> bool

Whether there are no live entries.

A stream can be empty and still exist, unlike every other collection here, because XADD followed by XDEL leaves a key whose last ID a new entry still has to beat.

pub fn last_id(&self) -> Id

The greatest ID ever appended, whether or not it is still here.

pub fn max_deleted_id(&self) -> Id

The greatest ID ever deleted, or Id::MIN if none ever was.

pub fn added(&self) -> u64

How many entries have ever been appended.

pub fn first_id(&self) -> Option<Id>

The lowest live ID, or None when there are none.

Walks the first node, because the first entry in it may have been deleted and its bytes are still there. That is at most a node’s worth of steps and it is only asked for by XINFO.

pub fn top_id(&self) -> Option<Id>

The highest live ID, or None when there are none.

Not the same as Stream::last_id, which is the last ID handed out and stays where it is when that entry is deleted. This is the greatest ID a reader can still find, and XSETID is the one caller that needs the difference, because it refuses to move the bookmark below an entry that is still there.

pub fn set_id( &mut self, last: Id, added: Option<u64>, max_deleted: Option<Id>, ) -> Result<(), Refused>

XSETID, which moves the bookmark and the two counters behind it.

The bookmark decides what the next XADD * hands out and what a group created at $ starts from, so moving it is how a replica is made to agree with a primary and how a stream is rebuilt from a log. The two counters are optional because Redis added them later and a caller that does not name them leaves them alone.

Errors

Refused::NotGreater when last is below an entry that is still in the stream, since a reader holding that entry’s ID would then be reading past the end of a stream that has not ended.

pub fn auto_id(&self, now: u64) -> Option<Id>

What the ID would be if XADD key * ran now with clock reading now.

The clock unless the clock has not moved, or has gone backwards, in which case it is the last ID with one added. A stream never goes back on its word about ordering just because the machine’s clock did.

pub fn auto_seq(&self, ms: u64) -> Option<Id>

The next sequence inside ms, which is what XADD key ms-* asks for.

pub fn append( &mut self, id: Id, fields: &[(&[u8], &[u8])], limits: Limits, ) -> Result<(), Refused>

Append an entry, which is XADD once the ID has been settled.

Errors

Refused when the ID is zero or is not greater than Stream::last_id. Nothing else can fail: an append never needs to move an entry that is already here.

pub fn delete(&mut self, id: Id) -> bool

Delete the entry with that ID, answering whether there was one.

The bytes stay where they are and a bit says the entry is gone, unless it was the last live entry in its node, in which case the node goes.

pub fn delete_ref(&mut self, id: Id, refs: Refs) -> Fate

Delete an entry, saying what to do about the groups, which is XDELEX.

Refs::Acked is the interesting one and it asks a wider question than its name does. An entry is safe to take when no group is holding it in a pending list and no group’s bookmark is still behind it, because a group that has not reached the entry yet has not had its chance at it. So a stream with one group sitting at 0-0 refuses every ACKED delete, and that is a real server’s answer and not an over careful reading of it.

pub fn ack_delete(&mut self, group: &[u8], id: Id, refs: Refs) -> (Fate, bool)

Acknowledge an entry for one group and then delete it, which is XACKDEL.

The acknowledgement is the part that decides the answer. An ID this group was not holding is Fate::Missing whether or not the entry is in the stream, and an ID it was holding is never Missing, so a caller reading the reply is being told about its own pending list and not about the log.

The flag beside the answer is whether an entry really left the log, which is not the same question. A group can be holding an ID that has since been deleted from under it, and acknowledging that one answers Fate::Gone because the pending list did lose it, while the log lost nothing. Only the flag is worth telling a keyspace subscriber about.

pub fn nack( &mut self, group: &[u8], id: Id, retry: Retry, force: bool, ) -> Option<bool>

Hand an entry back to a group without acknowledging it, which is XNACK.

force makes a pending entry out of one that was not pending, and like Stream::claim’s FORCE it only works on an entry that is really in the stream. Answers whether anything happened, and None when there is no such group.

pub fn lag(&self, group: &Group) -> Option<u64>

How far behind a group is, or None when that cannot be worked out.

Two ways of answering and the good one is tried first. If the group’s bookmark is somewhere the distance from the start of time is exactly known, which is the last ID, past it, or before the first entry left, that distance is the answer. Otherwise the group’s own counter will do, but only while nothing has been deleted at or above the bookmark, since a hole ahead of the group means it will read fewer entries than the subtraction is expecting.

Both paths and their order were read off Redis 8.10.1 rather than worked out, because reasoning gives the wrong answer on the case that matters: a group sitting at 0-0 on a stream trimmed from five entries to two reports a lag of two and not five, which is the estimate winning over a subtraction that is valid and is further from the truth.

pub fn trim_maxlen(&mut self, len: u64, exact: bool, limit: Option<u64>) -> u64

Cut the stream down to len entries, dropping the oldest, which is XTRIM key MAXLEN len. Answers how many went.

exact is Redis’s = against ~. Without it only whole nodes are dropped, so the stream is left at len or a little over and no node is ever rewritten. That is the form to use, and it is why ~ exists.

limit is Redis’s LIMIT, which stops the trim once that many entries have gone rather than once the stream is short enough. It exists because a capped stream that has fallen a long way behind would otherwise spend one command dropping millions of entries with the shard doing nothing else, and the next write will carry on where this one stopped.

pub fn trim_minid(&mut self, id: Id, exact: bool, limit: Option<u64>) -> u64

Drop every entry below id, which is XTRIM key MINID id. Answers how many went.

exact and limit mean what they do for Stream::trim_maxlen.

pub fn range<F>(&self, start: Id, end: Id, count: Option<usize>, f: F) -> usize

where F: FnMut(Id, Fields<'_>) -> bool,

Every live entry from start to end, both ends included, oldest first.

count stops the walk early, which is XRANGE ... COUNT n. The callback answers whether to carry on, so a caller filling a fixed reply can stop without knowing how many it wanted up front. Answers how many entries the callback saw.

pub fn rev_range<'s, F>( &'s self, start: Id, end: Id, count: Option<usize>, f: F, ) -> usize

where F: FnMut(Id, Fields<'_>) -> bool,

The same, newest first, which is XREVRANGE.

start and end are still the low and the high end of the range, so a caller does not have to swap them and the command layer does, once, where the argument order is Redis’s problem.

pub fn contains(&self, id: Id) -> bool

Whether an entry with this ID is there and live.

What XCLAIM asks before it hands a pending entry to somebody, since an entry that has been deleted or trimmed away is work nobody can do.

pub fn create_group(&mut self, name: &[u8], last: Id, read: Option<u64>) -> bool

Make a consumer group, and say whether it was not already there.

XGROUP CREATE. last is where it starts reading after, which is Stream::last_id for $ and Id::MIN for 0.

pub fn destroy_group(&mut self, name: &[u8]) -> bool

Take a group out, and say whether it was there.

pub fn group(&self, name: &[u8]) -> Option<&Group>

One group by name.

pub fn group_mut(&mut self, name: &[u8]) -> Option<&mut Group>

One group by name, to change.

pub fn groups(&self) -> impl Iterator<Item = (&[u8], &Group)> + '_

Every group, with its name.

pub fn read_group<F>( &mut self, group: &[u8], consumer: &[u8], count: Option<usize>, noack: bool, now: u64, f: F, ) -> Option<usize>

where F: FnMut(Id, Fields<'_>) -> bool,

Hand new entries to a consumer, which is XREADGROUP ... >.

Every entry after the group’s bookmark, up to count, delivered to consumer and written into the pending list as it goes. The consumer is created if it is not there, because a consumer exists by turning up.

noack is Redis’s NOACK, which hands the entries over without writing them into the pending list at all. The group still counts them as read, so the lag is the same either way, and the consumer is on its own if it dies holding one.

Answers how many entries the callback saw, or None when there is no such group.

pub fn read_group_pending<F>( &mut self, group: &[u8], consumer: &[u8], after: Id, count: Option<usize>, now: u64, f: F, ) -> Option<usize>

where F: FnMut(Id, Option<Fields<'_>>) -> bool,

Re-read what a consumer is already holding, which is XREADGROUP with an ID rather than >.

Every pending entry of that consumer after after, oldest first. Each one counts as handed out again, so its delivery time is reset and its count goes up. That is Redis’s behaviour, checked rather than assumed, and it is the right one: the count is how many times a consumer has been told to do this work, and a consumer re-reading its backlog after a restart has been told again.

An entry that has since been deleted or trimmed is still in the pending list and is handed to the callback with no fields, which is the null Redis puts in the reply. Clearing those out is Stream::claim’s job and not this one.

pub fn claim( &mut self, group: &[u8], consumer: &[u8], ids: &[Id], min_idle: u64, time: u64, retry: Option<u64>, bump: bool, force: bool, now: u64, gone: &mut Vec<Id>, ) -> Option<Vec<Id>>

Move pending entries to a consumer, which is XCLAIM.

Only entries idle at least min_idle move. time is what the delivery time becomes, retry replaces the delivery count when it is given, and bump says whether to add one to it, which JUSTID turns off. force makes a pending entry for an ID that is in the stream but was not pending.

An ID that is pending but no longer in the stream is dropped from the pending list rather than claimed, and reported through gone, which is what Redis does and what stops a deleted entry being handed round forever. Answers the IDs that moved.

pub fn autoclaim( &mut self, group: &[u8], consumer: &[u8], start: Id, min_idle: u64, count: usize, bump: bool, now: u64, gone: &mut Vec<Id>, ) -> Option<(Option<Id>, Vec<Id>)>

Sweep the pending list for stale entries and claim them, which is XAUTOCLAIM.

Starts at start and takes up to count entries that have been idle at least min_idle. Answers where a following call should carry on from, which is None at the end of the list, along with what was claimed and what was dropped for no longer being in the stream.

pub fn memory_bytes(&self) -> usize

How many bytes the entries and the groups take, not counting this struct.

The name is the one every other body in this crate uses, because the keyspace asks all of them the same question through one trait and a stream that answered it under a different name would need its own arm.

pub fn nodes(&self) -> usize

How many nodes there are, which only a test and XINFO STREAM FULL care about.

Trait Implementations

impl Bytes for Stream

fn memory_bytes(&self) -> usize

Bytes this value holds, not counting the slot it sits in.

impl Clone for Stream

fn clone(&self) -> Stream

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for Stream

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

Formats the value using the given formatter.

impl Default for Stream

fn default() -> Stream

Returns the “default value” for a type.

impl Eq for Stream

impl PartialEq for Stream

Two streams are the same when they hold the same entries and the same groups. The mutation count is not part of that. It is a number the resume cursors compare themselves against, nothing outside this file can see it, and a stream frozen and thawed is the same stream even though its count starts again at zero.

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

Equality operator ==.

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

Inequality operator !=.