Struct SubscriptionManager 

pub struct SubscriptionManager { /* private fields */ }

Manages active subscriptions and event routing.

The SubscriptionManager is the central hub for:

• Tracking active subscriptions per connection
• Receiving events from database listeners
• Matching events to subscriptions
• Broadcasting to transport adapters

Implementations

impl SubscriptionManager

pub fn new(schema: Arc<CompiledSchema>) -> Self

Create a new subscription manager.

Arguments

• schema - Compiled schema containing subscription definitions
• channel_capacity - Broadcast channel capacity (default: 1024)

pub fn with_capacity( schema: Arc<CompiledSchema>, channel_capacity: usize, ) -> Self

Create a new subscription manager with custom channel capacity.

pub fn receiver(&self) -> Receiver<SubscriptionPayload>

Get a receiver for subscription payloads.

Transport adapters use this to receive events for delivery.

pub fn subscribe( &self, subscription_name: &str, user_context: Value, variables: Value, connection_id: &str, ) -> Result<SubscriptionId, SubscriptionError>

Subscribe to a subscription type.

Arguments

• subscription_name - Name of the subscription type
• user_context - User authentication/authorization context
• variables - Runtime variables from client
• connection_id - Client connection identifier

Errors

Returns error if subscription not found or user not authorized.

pub fn subscribe_with_rls( &self, subscription_name: &str, user_context: Value, variables: Value, connection_id: &str, rls_conditions: Vec<(String, Value)>, ) -> Result<SubscriptionId, SubscriptionError>

Subscribe with pre-evaluated RLS conditions for event-level filtering.

The caller should evaluate the RLS policy at subscribe time and pass the resulting conditions (via extract_rls_conditions). During event delivery, each condition is checked against the event data: the event is only delivered if every condition matches.

Errors

Returns error if subscription not found or connection limit exceeded.

pub fn subscribe_planned( &self, plan: Arc<SubscriptionPlan>, user_context: Value, variables: Value, connection_id: &str, rls_conditions: Vec<(String, Value)>, ) -> Result<SubscriptionId, SubscriptionError>

Subscribe a subscriber whose events are served through plan (ruling AA 4): the executor planned it at subscribe time from the client’s selection and principal (Executor::plan_subscription). An event the plan suppresses is not delivered, and nothing tells the subscriber it existed.

Errors

As subscribe_with_rls.

pub fn subscription_arguments( &self, document: &ParsedQuery, variables: Option<&Value>, ) -> Result<Value>

The arguments document gives its root field, resolved against variables, for this manager’s schema: what a subscription registered here filters with (#1158).

Errors

As subscription_arguments: a document with no single root field, an unknown subscription or an undeclared argument.

pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), SubscriptionError>

Unsubscribe from a subscription.

Errors

Returns error if subscription not found.

pub fn unsubscribe_connection(&self, connection_id: &str)

Unsubscribe all subscriptions for a connection.

Called when a client disconnects.

Concurrency note

A concurrent subscribe call that runs between the DashMap entry removal and the per-subscription cleanup loop would create a new connection entry that is not cleaned up by this call. A second-pass removal after the first loop closes this window for all but the most extreme concurrent races. Any subscription that slips through is benign: it will receive events until the broadcast receiver is dropped (which happens on disconnect), and will be removed on the next disconnect or subscription-not-found event for that ID.

pub fn get_subscription(&self, id: SubscriptionId) -> Option<ActiveSubscription>

Get an active subscription by ID.

pub fn replace_plan( &self, id: SubscriptionId, plan: Arc<SubscriptionPlan>, ) -> Result<(), SubscriptionError>

Replace the plan every later event of subscription id is served through — a policy reload re-plans a live subscription (ruling AC 7).

Errors

SubscriptionError::NotActive when the subscription is gone.

pub fn update_rls_conditions( &self, id: SubscriptionId, rls_conditions: Vec<(String, Value)>, ) -> Result<(), SubscriptionError>

Replace the server-owned RLS conditions of a live subscription (#611).

Used when a schema hot-reload changes the row-visibility policy behind an already-connected subscription: the transport re-derives the conditions from the new policy and swaps them in, so the tightened boundary applies to the next delivered event without waiting for a reconnect. Delivery matching reads the conditions per event, so the swap takes effect immediately and atomically.

Errors

Returns SubscriptionError::NotActive when the subscription no longer exists (e.g. it raced a disconnect) — callers treat that as already gone.

pub fn get_connection_subscriptions( &self, connection_id: &str, ) -> Vec<ActiveSubscription>

Get all active subscriptions for a connection.

pub fn subscription_count(&self) -> usize

Get total number of active subscriptions.

pub fn connection_count(&self) -> usize

Get number of active connections with subscriptions.

pub fn publish_event(&self, event: SubscriptionEvent) -> usize

Publish an event to matching subscriptions.

This is called by the database listener when an event is received. The event is matched against all active subscriptions and delivered to matching ones.

Arguments

• event - The database event to publish

Returns

Number of subscriptions that matched the event.

Trait Implementations

impl Debug for SubscriptionManager

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.