ZenWebsocket.SubscriptionManager (ZenWebsocket v0.4.2)

Copy Markdown View Source

Manages subscription tracking for WebSocket connections.

Pure functional module - state ownership stays with Client GenServer. Tracks subscribed channels and provides restoration on reconnect.

Telemetry Events

The following telemetry events are emitted:

  • [:zen_websocket, :subscription_manager, :add] - Emitted when a channel is added.

    • Measurements: %{count: 1}
    • Metadata: %{channel: channel}

  • [:zen_websocket, :subscription_manager, :remove] - Emitted when a channel is removed.

    • Measurements: %{count: 1}
    • Metadata: %{channel: channel}

  • [:zen_websocket, :subscription_manager, :restore] - Emitted when subscriptions are restored.

    • Measurements: %{channel_count: integer()}
    • Metadata: %{channels: [String.t()]}

API Functions

FunctionArityDescriptionParam Kinds
handle_message2Handle incoming subscription confirmation messages.msg: value, state: value
build_restore_message1Build a restore message for reconnection.state: value
list1List all currently tracked subscriptions.state: value
remove2Remove a channel from the tracked subscription set.state: value, channel: value
add2Add a channel to the tracked subscription set.state: value, channel: value

Summary

Types

state()

Client state map containing subscription fields (subset of Client.state)

Functions

add(state, channel)

Adds a channel to the tracked subscription set.

build_restore_message(state)

Builds a restore message for reconnection.

handle_message(arg1, state)

Handles incoming subscription confirmation messages.

list(state)

Lists all currently tracked subscriptions.

remove(state, channel)

Removes a channel from the tracked subscription set.

Types

state()

@type state() :: %{
  :subscriptions => MapSet.t(String.t()),
  :config => %{:restore_subscriptions => boolean(), optional(atom()) => term()},
  optional(atom()) => term()
}

Client state map containing subscription fields (subset of Client.state)

Functions

add(state, channel)

@spec add(state(), String.t()) :: state()

Adds a channel to the tracked subscription set.

Called when a subscription confirmation is received.

build_restore_message(state)

@spec build_restore_message(state()) :: binary() | nil

Builds a restore message for reconnection.

Returns nil if:

  • No subscriptions to restore
  • restore_subscriptions config is false

Returns JSON-encoded subscribe message otherwise.

handle_message(arg1, state)

@spec handle_message(map(), state()) :: state()

Handles incoming subscription confirmation messages.

Extracts the channel from the message and adds it to tracked subscriptions.

list(state)

@spec list(state()) :: [String.t()]

Lists all currently tracked subscriptions.

remove(state, channel)

@spec remove(state(), String.t()) :: state()

Removes a channel from the tracked subscription set.

Called when unsubscribing from a channel.