Struct StreamTailer 

pub struct StreamTailer { /* private fields */ }

Stream tailer for a single peer’s CDC stream.

Implementations

impl StreamTailer

pub fn new( peer_id: String, stream_key: String, block_timeout: Duration, batch_size: usize, ) -> Self

Create a new stream tailer.

pub fn batch_size(&self) -> usize

Get the current batch size.

pub fn set_batch_size(&mut self, size: usize)

Update the batch size (for adaptive sizing).

pub async fn get_oldest_id( &self, conn: &mut ConnectionManager, ) -> Result<Option<String>>

Get the oldest entry ID in the stream, if any.

Returns None if the stream is empty or doesn’t exist.

pub async fn get_latest_id( &self, conn: &mut ConnectionManager, ) -> Result<Option<String>>

Get the latest (newest) entry ID in the stream, if any.

Returns None if the stream is empty or doesn’t exist.

pub async fn get_stream_length( &self, conn: &mut ConnectionManager, ) -> Result<u64>

Get the total number of entries in the stream.

Returns 0 if the stream doesn’t exist.

pub async fn check_cursor_valid( &self, conn: &mut ConnectionManager, cursor: &str, ) -> Result<Option<String>>

Check if a cursor is still valid (not older than the oldest stream entry).

Returns the oldest ID if the cursor is invalid, None if cursor is valid.

pub async fn read_events_checked( &self, conn: &mut ConnectionManager, cursor: &str, ) -> Result<ReadResult>

Read events from the stream with trim detection.

This is the preferred method for production use. It detects when the stream has been trimmed past our cursor and returns ReadResult::StreamTrimmed so callers can handle the gap appropriately.

pub async fn read_events( &self, conn: &mut ConnectionManager, cursor: &str, ) -> Result<Vec<CdcEvent>>

Read events from the stream starting after cursor.

Returns a vector of parsed events. Empty vector means no new events (timeout or stream empty).

The cursor should be the last successfully processed stream ID, or “0” to start from the beginning.

pub async fn read_events_range( &self, conn: &mut ConnectionManager, start: &str, count: usize, ) -> Result<Vec<CdcEvent>>

Read a range of events using XRANGE (non-blocking).

This is faster than XREAD for catchup scenarios because:

1. No blocking timeout overhead
2. Simpler command (no consumer group semantics)
3. Better for bulk reads when we know we’re behind

Use this when catching up, then switch to read_events (XREAD) when tailing.

Arguments

• start - Exclusive start ID (use “0” to start from beginning, or last processed ID)
• count - Maximum number of entries to fetch

Returns

Vector of parsed events. Empty means we’ve caught up.