Struct RedisMessage 

pub struct RedisMessage { /* private fields */ }

A Redis Streams delivery, read from a consumer group via XREADGROUP or XAUTOCLAIM.

Settlement follows the republish-retry model: ack is XACK; a retry re-appends a copy of the entry to the same stream and then acks the original (at-least-once, so a duplicate is possible if the process crashes between the two); a drop acks the original.

On a claiming subscription the retry is the read mode’s own: the entry stays in the pending entries list, and the subscription’s next read claims it back with the server’s delivery count one higher, which is the count this delivery reports.

Implementations

impl RedisMessage

pub fn id(&self) -> Option<&str>

The stream entry ID (for example 1700000000000-0) this message was read at.

pub const fn entry_id(&self) -> EntryId

The parsed entry id this message was read at.

pub fn group(&self) -> Option<&str>

The consumer group this delivery was read through, or None once the message has settled.

Trait Implementations

impl BuildBatchContext<RedisMessage> for StreamBatchContext

fn build(first: &RedisMessage) -> Self

Builds the context value by reading subscription-scoped fields out of the batch’s first delivery.

impl BuildContext<RedisMessage> for StreamContext

fn build(msg: &RedisMessage) -> Self

Builds the context value by reading fields out of msg.

impl BuildContext<RedisMessage> for PoolContext

fn build(msg: &RedisMessage) -> Self

Builds the context value by reading fields out of msg.

impl Debug for RedisMessage

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

Formats the value using the given formatter.

impl IncomingMessage for RedisMessage

fn redelivery_count(&self) -> Option<u64>

How many times the server has delivered this entry, counting this delivery.

Answered on the two read modes that claim, from the count Redis keeps in the pending entries list: claiming is told the count as of the previous delivery, so this one is added to it, and reclaim reads XPENDING after the claim, where this delivery is already counted. The fresh tail claims nothing and has no count of its own, so a cap on it is the runtime’s header instead.

fn supports_nack_after(&self) -> bool

Whether a delay reaches Redis rather than the runtime’s deferred copy.

True on a subscription that named a ZSET delay queue with RedisStream::delayed_retry, and on a claiming subscription, whose pending entries list is a delayed redelivery of Redis’s own. Elsewhere the runtime publishes the copy, which the mount site can customise with out_retry.

async fn nack_after(self, delay: Duration) -> Result<(), AckError>

Holds the message back for delay and redelivers it.

A ZSET delay queue serves the delay exactly: the delayed copy is ZADDed and the original XACKed, and the subscriber’s sweeper re-XADDs it once due, so the retry survives a crash. Without one, a claiming subscription leaves the entry pending and claims it back on its own min_idle, which is the granularity Redis offers there: a shorter delay waits min_idle, a longer one is honoured by the read that finds the entry still idle.

Errors

Returns AckError::Unsupported when the subscription has neither, or AckError::Broker when the ZADD or XACK fails.

fn payload(&self) -> &[u8]

Returns the raw payload of the message.

fn headers(&self) -> &HeaderMap

Returns the headers attached to the message.

fn partition_key(&self) -> Option<&[u8]>

Returns the routing key the broker partitioned this message by, or None when the message carries no key.

async fn ack(self) -> Result<(), AckError>

Acknowledges successful processing. Consumes the message handle.

async fn nack(self, requeue: bool) -> Result<(), AckError>

Negatively acknowledges the message. When requeue is true the broker should redeliver according to its own retry policy; when false it should drop or dead-letter the message.

impl Partitioned for RedisMessage

fn partition_key(&self) -> Option<&[u8]>

Returns the partition key for this item, or None if the broker should pick a partition.

impl Positioned for RedisMessage

The position of a delivery is the group cursor that redelivers it.

The cursor is exclusive (a group resumes after the id it holds), so the pinned position is the id immediately below this entry’s - seeking to it delivers this message again, followed by the entries after it. Repositioning is group-wide; see RedisGroupSeeker.

type Position = RedisGroupPosition

The position type, matching the subscription’s Seeker::Position.

fn position(&self) -> RedisGroupPosition

Returns the position of this delivery; seeking to it redelivers this message.