pub struct Group { /* private fields */ }Expand description
A consumer group over one stream.
Implementations§
Source§impl Group
impl Group
Sourcepub fn new(last: Id, read: Option<u64>) -> Group
pub fn new(last: Id, read: Option<u64>) -> Group
A group reading after last, having read read entries.
Sourcepub fn entries_read(&self) -> Option<u64>
pub fn entries_read(&self) -> Option<u64>
How many entries the group has read, when that is known.
Sourcepub fn set_id(&mut self, last: Id, read: Option<u64>)
pub fn set_id(&mut self, last: Id, read: Option<u64>)
Move the bookmark, which is XGROUP SETID.
The PEL is left alone, because the entries in it were handed to somebody who has not finished and moving the bookmark says nothing about them.
Sourcepub fn pending_len(&self) -> usize
pub fn pending_len(&self) -> usize
How many entries are pending across the whole group.
Sourcepub fn nacked_len(&self) -> usize
pub fn nacked_len(&self) -> usize
How many of those nobody is holding, which XNACK is what makes nonzero.
Sourcepub fn pending_all(&self) -> impl Iterator<Item = (Id, &Nack)> + '_
pub fn pending_all(&self) -> impl Iterator<Item = (Id, &Nack)> + '_
Every pending entry with what is known about it, oldest first.
The whole ledger and no filter, which is what an RDB payload carries and
what nothing on the wire ever asks for, since XPENDING always has a
range and usually a count.
Sourcepub fn pending_bounds(&self) -> Option<(Id, Id)>
pub fn pending_bounds(&self) -> Option<(Id, Id)>
The lowest and highest pending IDs, which is the XPENDING summary.
Sourcepub fn consumer_named(&self, name: &[u8]) -> Option<&Consumer>
pub fn consumer_named(&self, name: &[u8]) -> Option<&Consumer>
A consumer by name.
Sourcepub fn consumers(&self) -> impl Iterator<Item = &Consumer> + '_
pub fn consumers(&self) -> impl Iterator<Item = &Consumer> + '_
Every consumer, in the order they were created.
Sourcepub fn consumer_or_create(&mut self, name: &[u8], now: u64) -> u32
pub fn consumer_or_create(&mut self, name: &[u8], now: u64) -> u32
The slot for name, making the consumer if it is not there yet.
This is what XREADGROUP does, since a consumer exists because it turned
up rather than because anybody declared it.
Sourcepub fn create_consumer(&mut self, name: &[u8], now: u64) -> bool
pub fn create_consumer(&mut self, name: &[u8], now: u64) -> bool
Make a consumer and say whether it was not already there.
XGROUP CREATECONSUMER, which answers 1 when it made one.
Sourcepub fn delete_consumer(&mut self, name: &[u8]) -> u64
pub fn delete_consumer(&mut self, name: &[u8]) -> u64
Take a consumer out and say how many entries it was holding.
Those entries stop being pending at all, which is Redis’s behaviour and is the point of the command: deleting a consumer is how you give up on the work it was holding when you would rather lose it than claim it.
Sourcepub fn touch(&mut self, slot: u32, now: u64, read: bool)
pub fn touch(&mut self, slot: u32, now: u64, read: bool)
Mark that a consumer was heard from.
read says whether it got anything, which is what separates seen from
active.
Sourcepub fn deliver(&mut self, slot: u32, id: Id, now: u64) -> bool
pub fn deliver(&mut self, slot: u32, id: Id, now: u64) -> bool
Hand an entry to a consumer for the first time.
The bookmark moves, since this is the > path and the entry is new to
the group. Answers false if the slot is empty, which a caller that got
its slot from Group::consumer_or_create cannot hit.
Sourcepub fn skip(&mut self, id: Id)
pub fn skip(&mut self, id: Id)
Move the bookmark past an entry without writing it into the ledger.
This is XREADGROUP ... NOACK, which is a consumer saying it does not
want the work tracked. The group still counts the entry as read, because
the lag is about how far behind the group is and not about how much of
it is outstanding, so a NOACK reader that has caught up reports a lag of
zero the same as any other.
Sourcepub fn redeliver(&mut self, id: Id, now: u64) -> bool
pub fn redeliver(&mut self, id: Id, now: u64) -> bool
Hand an entry to whoever already holds it, which is a history read.
XREADGROUP with an ID rather than > is a consumer asking for what it
was already given, and Redis treats that as a real delivery: the time is
reset and the count goes up, exactly as if the entry had been handed out
again. Checked against Redis 8.10.1, where a history read of an entry
idle for 2006 milliseconds left it idle for 2 with its count up by one.
It reads as surprising until you think about what the count is for. It counts 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.
Sourcepub fn set_read(&mut self, read: Option<u64>)
pub fn set_read(&mut self, read: Option<u64>)
Put the read counter where the stream has worked out it belongs.
The counter is a fact about the stream and not about the group, since
what a delivery does to it depends on whether anything has been deleted
ahead of the group. crate::stream::Stream::read_group is the one
caller, and it is the one that can see both.
Sourcepub fn ack(&mut self, id: Id) -> bool
pub fn ack(&mut self, id: Id) -> bool
Finish with an entry, which is XACK.
Answers whether it was pending. Acknowledging something twice is not an error, it just does nothing the second time, because a consumer that crashed between doing the work and sending the ack will send it again.
Sourcepub fn release(&mut self, id: Id, retry: Retry) -> bool
pub fn release(&mut self, id: Id, retry: Retry) -> bool
Hand an entry back to the group without acknowledging it, which is XNACK.
The entry stays pending and stops belonging to anybody, so it reads as
idle for longer than any claim can ask for and the next XAUTOCLAIM
takes it. retry is what the delivery count becomes, which is the whole
difference between the three words XNACK takes.
Answers whether it was pending. The bookmark does not move, so a > read
will not hand it out again: releasing an entry offers it to a claim and
not to the group’s next reader, which is Redis’s behaviour and the only
one that keeps a released entry from being delivered twice over.
Sourcepub fn force_release(&mut self, id: Id, retry: Retry)
pub fn force_release(&mut self, id: Id, retry: Retry)
Make a released entry out of one that was not pending, which is
XNACK ... FORCE.
The caller has to have checked that the entry is really in the stream,
for the same reason Group::force does. A count of zero is where a
released entry that has never been delivered starts, whatever word was
used, because there is no earlier count for FAIL to keep.
Sourcepub fn forget(&mut self, id: Id) -> bool
pub fn forget(&mut self, id: Id) -> bool
Drop a pending entry without it having been acknowledged.
What happens to a NACK whose entry is no longer in the stream. XCLAIM
and XAUTOCLAIM both clear those out as they find them, because a
pending entry nobody can ever read is work no consumer can ever finish.
Sourcepub fn claim(
&mut self,
id: Id,
slot: u32,
time: u64,
count: Option<u64>,
bump: bool,
) -> bool
pub fn claim( &mut self, id: Id, slot: u32, time: u64, count: Option<u64>, bump: bool, ) -> bool
Move an entry to another consumer, which is the middle of XCLAIM.
time is when it should count as having been handed out, which is now
for a plain claim and something a caller chose for IDLE or TIME.
count replaces the delivery count when it is given, which is
RETRYCOUNT, and otherwise the count goes up by one unless bump says
not to, which is JUSTID.
Answers false when the entry was not pending or the slot is empty.
Sourcepub fn force(&mut self, id: Id, slot: u32, time: u64, count: u64) -> bool
pub fn force(&mut self, id: Id, slot: u32, time: u64, count: u64) -> bool
Make a pending entry that was not pending, which is XCLAIM FORCE.
The caller has to have checked that the entry is really in the stream,
because this cannot see the stream and creating a NACK for an entry that
is not there is exactly the state Group::forget exists to clean up.
Sourcepub fn pending_range<F>(&self, want: Filter, now: u64, f: F) -> usize
pub fn pending_range<F>(&self, want: Filter, now: u64, f: F) -> usize
Pending entries in want, oldest first.
The callback answers whether to carry on, and gets None for the owner
of an entry that has been released, which XPENDING writes as an empty
name. A consumer filter never matches one of those, since asking what a
named consumer is holding is asking about entries that have an owner.
Sourcepub fn pending_counts(&self) -> impl Iterator<Item = (&[u8], usize)> + '_
pub fn pending_counts(&self) -> impl Iterator<Item = (&[u8], usize)> + '_
How many entries each consumer is holding, for the XPENDING summary.
Sourcepub fn claimable(
&self,
start: Id,
min_idle: u64,
now: u64,
limit: usize,
out: &mut Vec<Id>,
) -> Option<Id>
pub fn claimable( &self, start: Id, min_idle: u64, now: u64, limit: usize, out: &mut Vec<Id>, ) -> Option<Id>
The IDs an XAUTOCLAIM would take, from start and idle at least
min_idle, and where a following call should carry on from.
Only the scan, because deciding what to do with each one needs the
stream and this does not have it. The cursor is None when the scan
reached the end, which is the 0-0 Redis answers with.
Sourcepub fn memory_bytes(&self) -> usize
pub fn memory_bytes(&self) -> usize
How many bytes this group takes, not counting the struct itself.
The pending map is counted per entry at the size of a key and a value
plus a share of the node around them, rather than exactly, because a
BTreeMap does not say how many nodes it has and the answer is only
ever read by MEMORY USAGE and the eviction total. A B-tree node here
holds eleven entries and some overhead, and a sixteenth of an entry is
close enough for both.
Trait Implementations§
impl Eq for Group
Source§impl PartialEq for Group
Two groups are the same when they hold the same entries for the same
consumers at the same place. The resume cursor is not part of that. It is a
note about where a walk got to in a blob, a freeze and thaw drops it, and a
group that has read something is not a different group from the same group
before it did.
impl PartialEq for Group
Two groups are the same when they hold the same entries for the same consumers at the same place. The resume cursor is not part of that. It is a note about where a walk got to in a blob, a freeze and thaw drops it, and a group that has read something is not a different group from the same group before it did.