Struct MessagesManager 

pub struct MessagesManager {
    messages_service: Arc<MessagesService>,
    broadcast_tx: Arc<Sender<StreamMessage>>,
    catch_up_tx: Arc<Sender<CatchUpResponse>>,
    message_events_tx: Arc<Sender<MessageEvent>>,
    _broadcast_keeper: InactiveReceiver<StreamMessage>,
    _catch_up_keeper: InactiveReceiver<CatchUpResponse>,
    _message_events_keeper: InactiveReceiver<MessageEvent>,
    state: SharedObservable<SubscriptionState, AsyncLock>,
    _sync_handler: Arc<AbortOnDrop<Result<()>>>,
    catch_up_request_id: Arc<AtomicU32>,
}

High-level messages manager that provides a unified interface for message operations.

The MessagesManager combines message broadcasting and subscription management:

• Message Broadcasting: Distributes incoming messages to multiple subscribers
• Subscription Management: Manages server-side subscriptions with in-flight updates
• Stream Filtering: Provides client-side filtering and routing capabilities
• Lifecycle Management: Automatic subscription creation, updates, and cleanup

This is the primary interface for interacting with the messaging system.

Fields

messages_service: Arc<MessagesService>

The underlying messages service for RPC operations

broadcast_tx: Arc<Sender<StreamMessage>>

Broadcast sender for distributing messages to subscribers

catch_up_tx: Arc<Sender<CatchUpResponse>>

Broadcast sender for distributing catch-up responses to subscribers

message_events_tx: Arc<Sender<MessageEvent>>

Broadcast sender for all message events (for persistence and monitoring)

_broadcast_keeper: InactiveReceiver<StreamMessage>

Keeper receiver to prevent broadcast channel closure (not actively used)

_catch_up_keeper: InactiveReceiver<CatchUpResponse>

Keeper receiver to prevent catch-up channel closure (not actively used)

_message_events_keeper: InactiveReceiver<MessageEvent>

Keeper receiver to prevent message events channel closure (not actively used)

state: SharedObservable<SubscriptionState, AsyncLock>

Current subscription state (persistent across reconnections)

_sync_handler: Arc<AbortOnDrop<Result<()>>>

Background task handle for syncing with the server

catch_up_request_id: Arc<AtomicU32>

Catch-up request ID counter

Implementations

impl MessagesManager

pub fn builder() -> MessagesManagerBuilder

fn safe_broadcast<T: Clone>(sender: &Sender<T>, message: T, context: &str)

Helper function to safely broadcast messages using try_broadcast Handles TrySendError cases gracefully without panicking

fn new_with_state( messages_service: MessagesService, messages_stream: MessagesStream, catch_up_stream: CatchUpStream, state: SubscriptionState, buffer_size: Option<usize>, ) -> Self

Create a new MessagesManager with existing subscription state.

This allows restoring a manager to a previous state after reconnection.

Arguments

• messages_service - The underlying messages service for RPC operations
• messages_stream - The stream of messages from the server
• state - Previous subscription state to restore
• buffer_size - Optional buffer size for the broadcast channel (default: 1000)

pub async fn subscribe(&self) -> Result<()>

pub async fn ensure_contains_filter(&self, filter: Filter) -> Result<()>

pub async fn catch_up_and_subscribe( self, filter: Filter, since: Option<String>, ) -> Result<impl Stream<Item = Box<MessageFull>>>

pub fn filtered_messages_stream( self, filter: Filter, ) -> Pin<Box<dyn Stream<Item = Box<MessageFull>> + Send>>

pub fn filtered_fn<F, T>(self, filter: F) -> impl Stream<Item = T>

where F: Fn(StreamMessage) -> Option<T> + Send + Clone + 'static,

Get a filtered stream of messages.

This creates a client-side filtered stream from the internal broadcast channel. The filter function is applied to all messages received by the manager.

Arguments

• filter - A function that returns true for messages to include

Returns

A stream of messages that match the filter

pub fn all_messages_stream(&self) -> Receiver<StreamMessage>

Get a stream of all messages.

Returns

A stream of all messages received by the manager

pub async fn publish(&self, message: MessageFull) -> Result<PublishResult>

pub async fn get_subscription_state(&self) -> SubscriptionState

Get the current subscription state for persistence.

This state can be saved and later restored using MessagesManagerBuilder.

Returns

A clone of the current subscription state

pub async fn get_latest_stream_height(&self) -> Option<String>

Get the latest stream height received.

This can be used to determine how up-to-date the client is.

pub async fn get_current_filters(&self) -> MessageFilters

Get the current combined filters.

This shows all the filters that are currently active in the subscription.

pub fn message_events_stream(&self) -> Receiver<MessageEvent>

Get a stream of all message events for persistence and monitoring.

This stream captures every message activity including:

• Messages received from subscriptions
• Messages sent by this client
• Historical messages from catch-up requests
• Stream height updates
• Catch-up completion notifications

This is primarily intended for persistence services and audit logging.

Returns

A stream of MessageEvent that captures all message activities

Trait Implementations

impl Clone for MessagesManager

fn clone(&self) -> MessagesManager

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl MessagesManagerTrait for MessagesManager

fn message_events_stream(&self) -> Receiver<MessageEvent>

Get a stream of all message events for persistence and monitoring

fn get_subscription_state_updates<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Subscriber<SubscriptionState, AsyncLock>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to subscription state changes for reactive programming

fn subscribe<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to messages with current filters

fn publish<'life0, 'async_trait>( &'life0 self, message: MessageFull, ) -> Pin<Box<dyn Future<Output = Result<PublishResult>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Publish a message

fn ensure_contains_filter<'life0, 'async_trait>( &'life0 self, filter: Filter, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Ensure a filter is included in the subscription

fn messages_stream(&self) -> Receiver<StreamMessage>

Get a stream of incoming messages

fn catch_up_stream(&self) -> Receiver<CatchUpResponse>

Get a stream of catch-up responses

fn filtered_messages_stream( &self, filter: Filter, ) -> Pin<Box<dyn Stream<Item = Box<MessageFull>> + Send>>

Get a filtered stream of messages matching the given filter

fn catch_up_and_subscribe<'life0, 'async_trait>( &'life0 self, filter: Filter, since: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Box<MessageFull>> + Send>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Catch up to historical messages and subscribe to new ones for a filter

fn user_data<'life0, 'async_trait>( &'life0 self, author: KeyId, storage_key: StoreKey, ) -> Pin<Box<dyn Future<Output = Result<Option<MessageFull>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get user data by author and storage key (for PQXDH inbox fetching)

fn check_messages<'life0, 'async_trait>( &'life0 self, message_ids: Vec<MessageId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Option<String>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Check which messages the server already has and return their global stream IDs. Returns a vec of Option<String> in the same order as the input, where: