ZenWebsocket.SubscriptionManager (ZenWebsocket v0.7.1)

Copy Markdown View Source

Tracks Deribit-dialect JSON-RPC subscriptions for reconnect restoration.

Only public/subscribe and public/unsubscribe with params.channels are auto-tracked (plus their id-keyed confirmations and rejections). Other venues' subscribe shapes are ignored and are not automatically restored; build_restore_message/1 always emits Deribit's payload.

Client.subscribe/2 sends no JSON-RPC id, so channels are recorded at send time. A server-side rejection leaves them in the restore set. Id-carrying requests wait for a result or error.

Pure functional module — state ownership stays with Client GenServer.

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_message2Track Deribit public/subscribe and public/unsubscribe requests and confirmations.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(msg, state)

Tracks Deribit public/subscribe and public/unsubscribe requests.

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.

Used after a subscribe operation is confirmed or when it has no correlation ID.

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(msg, state)

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

Tracks Deribit public/subscribe and public/unsubscribe requests.

Requests with no id (including Client.subscribe/2) apply immediately. Id-carrying requests wait for a result or error. Data ticks and non-Deribit subscribe shapes are ignored.

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.

Used after an unsubscribe operation is confirmed or when it has no correlation ID.