Struct SubscriptionManager 

pub struct SubscriptionManager { /* private fields */ }

A manager for SierraDB subscriptions that handles a shared push channel and demultiplexes messages to individual subscriptions.

Implementations

impl SubscriptionManager

pub async fn new(client: &Client) -> RedisResult<Self>

Create a new SubscriptionManager with the given Redis client.

This establishes a multiplexed connection and starts a background task to process push messages.

pub async fn subscribe_to_stream<S: ToRedisArgs>( &mut self, stream_id: S, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream.

pub async fn subscribe_to_stream_with_window<S: ToRedisArgs>( &mut self, stream_id: S, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream with optional windowing.

pub async fn subscribe_to_stream_from_version<S: ToRedisArgs>( &mut self, stream_id: S, from_version: u64, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream starting from a specific version.

pub async fn subscribe_to_stream_from_version_with_window<S: ToRedisArgs>( &mut self, stream_id: S, from_version: u64, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream starting from a specific version with windowing.

pub async fn subscribe_to_partition( &mut self, partition: u16, ) -> RedisResult<EventSubscription>

Subscribe to events from a specific partition.

pub async fn subscribe_to_partition_with_window( &mut self, partition: u16, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a specific partition with windowing.

pub async fn subscribe_to_partition_key( &mut self, key: Uuid, ) -> RedisResult<EventSubscription>

Subscribe to events from a partition identified by a UUID key.

pub async fn subscribe_to_partition_key_with_window( &mut self, key: Uuid, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a partition identified by a UUID key with windowing.

pub async fn subscribe_to_stream_with_partition_key<S: ToRedisArgs>( &mut self, stream_id: S, partition_key: Uuid, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream with a partition key.

pub async fn subscribe_to_stream_with_partition_key_and_window<S: ToRedisArgs>( &mut self, stream_id: S, partition_key: Uuid, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream with a partition key and windowing.

pub async fn subscribe_to_stream_with_partition_and_version<S: ToRedisArgs>( &mut self, stream_id: S, partition_key: Uuid, from_version: u64, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream with both partition key and starting version.

pub async fn subscribe_to_stream_with_partition_and_version_and_window<S: ToRedisArgs>( &mut self, stream_id: S, partition_key: Uuid, from_version: u64, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream with partition key, starting version, and windowing.

pub async fn subscribe_to_partition_from_sequence( &mut self, partition: u16, from_sequence: u64, ) -> RedisResult<EventSubscription>

Subscribe to events from a partition starting from a specific sequence.

pub async fn subscribe_to_partition_from_sequence_with_window( &mut self, partition: u16, from_sequence: u64, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a partition starting from a specific sequence with windowing.

pub async fn subscribe_to_partition_key_from_sequence( &mut self, key: Uuid, from_sequence: u64, ) -> RedisResult<EventSubscription>

Subscribe to events from a partition (by key) starting from a specific sequence.

pub async fn subscribe_to_partition_key_from_sequence_with_window( &mut self, key: Uuid, from_sequence: u64, window_size: u32, ) -> RedisResult<EventSubscription>

Subscribe to events from a partition (by key) starting from a specific sequence with windowing.

pub async fn acknowledge_up_to_cursor( &mut self, subscription_id: Uuid, cursor: u64, ) -> RedisResult<()>

Acknowledge events up to a specific cursor for a subscription.

The cursor value is provided with each event in the subscription stream. Acknowledging up to a cursor allows the subscription window to advance.

pub async fn subscribe_to_partitions( &mut self, partition_range: &str, from_sequence: u64, window_size: Option<u32>, ) -> RedisResult<EventSubscription>

Subscribe to events from multiple partitions with the same starting sequence.

Arguments

• partition_range - String specifying partitions: “*”, “0-127”, “0,1,5”, or “42”
• from_sequence - Starting sequence number for all partitions
• window_size - Optional window size for flow control

Examples

let sub = manager.subscribe_to_partitions("*", 1000, Some(100)).await?;
let sub = manager.subscribe_to_partitions("0-127", 500, None).await?;
let sub = manager.subscribe_to_partitions("0,1,5", 0, Some(50)).await?;

pub async fn subscribe_to_partitions_with_sequences( &mut self, partition_sequences: HashMap<u16, u64>, window_size: Option<u32>, ) -> RedisResult<EventSubscription>

Subscribe to events from multiple partitions with per-partition sequences.

Arguments

• partition_sequences - Map of partition_id -> from_sequence
• window_size - Optional window size for flow control

Example

let mut sequences = HashMap::new();
sequences.insert(0, 500);
sequences.insert(1, 1200);
sequences.insert(127, 999);
let sub = manager.subscribe_to_partitions_with_sequences(sequences, Some(100)).await?;

pub async fn subscribe_to_all_partitions( &mut self, from_sequence: u64, window_size: Option<u32>, ) -> RedisResult<EventSubscription>

Subscribe to all partitions starting from the same sequence.

This is a convenience method equivalent to subscribe_to_partitions("*", from_sequence, window_size).

pub async fn subscribe_to_partition_range( &mut self, start_partition: u16, end_partition: u16, from_sequence: u64, window_size: Option<u32>, ) -> RedisResult<EventSubscription>

Subscribe to a range of partitions starting from the same sequence.

Arguments

• start_partition - Starting partition ID (inclusive)
• end_partition - Ending partition ID (inclusive)
• from_sequence - Starting sequence number for all partitions in range
• window_size - Optional window size for flow control

Example

// Subscribe to partitions 0-127 starting from sequence 1000
let sub = manager.subscribe_to_partition_range(0, 127, 1000, Some(50)).await?;

pub async fn subscribe_to_stream_from_latest<S: ToRedisArgs>( &mut self, stream_id: S, ) -> RedisResult<EventSubscription>

Subscribe to events from a stream starting from latest.

pub async fn subscribe_to_all_partitions_from_latest( &mut self, ) -> RedisResult<EventSubscription>

Subscribe to events from all partitions starting from latest.

pub async fn subscribe_to_all_partitions_with_fallback( &mut self, partition_sequences: HashMap<u16, u64>, fallback_sequence: u64, window_size: Option<u32>, ) -> RedisResult<EventSubscription>

Subscribe to events from all partitions with custom starting sequences for specific partitions and a fallback.

Arguments

• partition_sequences - Map of partition_id -> starting_sequence for specific partitions
• fallback_sequence - Starting sequence for all other partitions not specified in the map
• window_size - Optional window size for flow control

Example

let mut sequences = HashMap::new();
sequences.insert(0, 1000);    // Partition 0 starts from sequence 1000
sequences.insert(5, 2500);    // Partition 5 starts from sequence 2500
// All other partitions start from sequence 100 (fallback)
let sub = manager.subscribe_to_all_partitions_with_fallback(sequences, 100, Some(50)).await?;

pub async fn subscribe_to_all_partitions_flexible( &mut self, from_map: HashMap<u16, u64>, fallback_sequence: Option<u64>, window_size: Option<u32>, ) -> RedisResult<EventSubscription>

Subscribe to events from all partitions with flexible starting position options.

Arguments

• from_map - Map of partition_id -> starting_sequence for specific partitions (empty map means no overrides)
• fallback_sequence - Optional fallback sequence for partitions not in the map
• window_size - Optional window size for flow control

Examples

// Subscribe from latest on all partitions
let sub = manager.subscribe_to_all_partitions_flexible(HashMap::new(), None, Some(50)).await?;

// Subscribe with specific sequences for some partitions, latest for others
let mut sequences = HashMap::new();
sequences.insert(0, 1000);
sequences.insert(5, 2500);
let sub = manager.subscribe_to_all_partitions_flexible(sequences, None, Some(50)).await?;

// Subscribe with fallback for all partitions not specified
let sub = manager.subscribe_to_all_partitions_flexible(sequences, Some(100), Some(50)).await?;

Trait Implementations

impl Clone for SubscriptionManager

fn clone(&self) -> SubscriptionManager

Returns a duplicate of the value.

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

Performs copy-assignment from source.