Struct ClientManager 

pub struct ClientManager { /* private fields */ }

Manages all connected WebSocket clients using lock-free DashMap.

Key design decisions:

• Uses DashMap for lock-free concurrent access to client registry
• Uses try_send instead of send to never block on slow clients
• Disconnects clients that are backpressured (queue full) to prevent cascade failures
• All public methods are non-blocking or use fine-grained per-key locks
• Supports configurable rate limiting per IP, subject, and global defaults

Implementations

impl ClientManager

pub fn new() -> Self

pub fn with_config(config: RateLimitConfig) -> Self

Create a new ClientManager with the given rate limit configuration

pub fn from_env() -> Self

Load configuration from environment variables

See RateLimitConfig::from_env for supported variables.

pub fn with_timeout(self, timeout: Duration) -> Self

Set the client timeout for stale client cleanup

pub fn with_message_queue_size(self, queue_size: usize) -> Self

Set the message queue size per client

pub fn with_max_connections_per_ip(self, max: usize) -> Self

Set a global limit on connections per IP address

pub fn with_rate_limit_window(self, window: Duration) -> Self

Set the rate limit window duration

pub fn with_default_limits(self, limits: Limits) -> Self

Set default limits applied when auth token doesn’t specify limits

These limits act as server-wide fallbacks for connections where the authentication token doesn’t include explicit limits.

pub fn with_rate_limiter(self, rate_limiter: Arc<WebSocketRateLimiter>) -> Self

Set a WebSocket rate limiter for granular rate control

pub fn rate_limiter(&self) -> Option<&WebSocketRateLimiter>

Get the rate limiter if configured

pub fn rate_limit_config(&self) -> &RateLimitConfig

Get the current rate limit configuration

pub fn add_client( &self, client_id: Uuid, ws_sender: WebSocketSender, auth_context: Option<AuthContext>, remote_addr: SocketAddr, )

Add a new client connection.

Spawns a dedicated sender task for this client that reads from its mpsc channel and writes to the WebSocket. If the WebSocket write fails, the client is automatically removed from the registry.

pub fn remove_client(&self, client_id: Uuid)

Remove a client from the registry.

pub fn update_client_auth( &self, client_id: Uuid, auth_context: AuthContext, ) -> bool

Update the auth context for a client.

Used for in-band auth refresh without reconnecting.

pub fn check_and_remove_expired(&self, client_id: Uuid) -> bool

Check if a client’s token has expired.

Returns true if the client has an auth context and it has expired. If expired, the client is removed from the registry.

pub fn client_count(&self) -> usize

Get the current number of connected clients.

This is lock-free and returns an approximate count (may be slightly stale under high concurrency, which is fine for max_clients checks).

pub fn send_to_client( &self, client_id: Uuid, data: Arc<Bytes>, ) -> Result<(), SendError>

Send data to a specific client (non-blocking).

This method NEVER blocks. If the client’s queue is full, the client is considered too slow and is disconnected to prevent cascade failures. Use this for live streaming updates.

For initial snapshots where you expect to send many messages at once, use send_to_client_async instead which will wait for queue space.

pub async fn send_to_client_async( &self, client_id: Uuid, data: Arc<Bytes>, ) -> Result<(), SendError>

Send data to a specific client (async, waits for queue space).

This method will wait if the client’s queue is full, allowing the client time to catch up. Use this for initial snapshots where you need to send many messages at once.

For live streaming updates, use send_to_client instead which will disconnect slow clients rather than blocking.

pub async fn send_text_to_client( &self, client_id: Uuid, text: String, ) -> Result<(), SendError>

Send a text message to a specific client (async).

This method sends a text message directly to the client’s WebSocket. Used for control messages like auth refresh responses.

pub async fn send_compressed_async( &self, client_id: Uuid, payload: CompressedPayload, ) -> Result<(), SendError>

Send a potentially compressed payload to a client (async).

Compressed payloads are sent as binary frames (raw gzip). Uncompressed payloads are sent as text frames (JSON).

pub fn update_subscription( &self, client_id: Uuid, subscription: Subscription, ) -> bool

Update the subscription for a client.

pub fn update_client_last_seen(&self, client_id: Uuid)

Update the last_seen timestamp for a client.

pub fn check_inbound_message_allowed( &self, client_id: Uuid, ) -> Result<(), AuthDeny>

Check whether an inbound message is allowed for a client.

pub fn get_subscription(&self, client_id: Uuid) -> Option<Subscription>

Get the subscription for a client.

pub fn has_client(&self, client_id: Uuid) -> bool

Check if a client exists.

pub async fn add_client_subscription( &self, client_id: Uuid, sub_key: String, token: CancellationToken, ) -> bool

pub async fn remove_client_subscription( &self, client_id: Uuid, sub_key: &str, ) -> bool

pub async fn cancel_all_client_subscriptions(&self, client_id: Uuid)

pub fn cleanup_stale_clients(&self) -> usize

Remove stale clients that haven’t been seen within the timeout period.

pub fn start_cleanup_task(&self)

Start a background task that periodically cleans up stale clients.

pub async fn check_connection_allowed( &self, remote_addr: SocketAddr, auth_context: &Option<AuthContext>, ) -> Result<(), AuthDeny>

ENFORCEMENT HOOKS

These methods provide hooks for enforcing limits based on auth context. They check limits before allowing operations and return errors if limits are exceeded. Check if a connection is allowed for the given auth context.

Returns Ok(()) if the connection is allowed, or an error with a reason if not.

pub async fn check_subscription_allowed( &self, client_id: Uuid, ) -> Result<(), AuthDeny>

Check if a subscription is allowed for the given client.

Returns Ok(()) if the subscription is allowed, or an error with a reason if not.

pub fn get_metering_key(&self, client_id: Uuid) -> Option<String>

Get metering key for a client

pub fn get_auth_context(&self, client_id: Uuid) -> Option<AuthContext>

Get auth context for a client.

pub fn check_snapshot_allowed( &self, client_id: Uuid, requested_rows: u32, ) -> Result<(), AuthDeny>

Check if a snapshot request is allowed (based on max_snapshot_rows limit)

Uses token limits if available, falls back to default limits from RateLimitConfig.

Trait Implementations

impl Clone for ClientManager

fn clone(&self) -> ClientManager

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Default for ClientManager

fn default() -> Self

Returns the “default value” for a type.