Struct ConsumerRegistry 

pub struct ConsumerRegistry { /* private fields */ }

Registry that manages durable consumers with persistent cursor positions.

Consumer state is stored as system events in the WAL pipeline so it survives Core restarts. In-memory state is held in a DashMap for O(1) access.

Two modes:

• In-memory (new()): no persistence, for tests and backward compatibility
• Durable (new_durable()): backed by SystemMetadataStore, survives restarts

Implementations

impl ConsumerRegistry

pub fn new() -> Self

Create an in-memory-only consumer registry (no persistence).

pub fn new_durable(system_store: Arc<SystemMetadataStore>) -> Self

Create a durable consumer registry backed by SystemMetadataStore.

On construction, replays all _system.consumer.* events to rebuild the in-memory cache from WAL.

pub fn register( &self, consumer_id: &str, event_type_filters: &[String], ) -> Consumer

Register a consumer (or update its filters if it already exists).

pub fn get(&self, consumer_id: &str) -> Option<Consumer>

Get a consumer by ID. Returns None if not registered.

pub fn get_or_create(&self, consumer_id: &str) -> Consumer

Get or implicitly create a consumer.

pub fn ack( &self, consumer_id: &str, position: u64, max_offset: u64, ) -> Result<(), String>

Acknowledge events up to a given global offset. Returns Ok(()) on success, Err if the position is beyond the max offset.

pub fn restore(&self, consumer: Consumer)

Restore consumer state (called during WAL recovery).

pub fn count(&self) -> usize

Number of registered consumers.

pub fn matches_filters(event_type: &str, filters: &[String]) -> bool

Check if an event type matches a consumer’s filters. Empty filters = match all.

Trait Implementations

impl Default for ConsumerRegistry

fn default() -> Self

Returns the “default value” for a type.