Current section

Files

Jump to
phoenix_socket_client lib phoenix socket_client agent.ex
Raw

lib/phoenix/socket_client/agent.ex

defmodule Phoenix.SocketClient.Agent do
@moduledoc """
Agent-based state management for WebSocket connection configuration and status.
This module provides centralized state management for socket connections,
including configuration parameters, connection status, and custom state values.
All state is stored in an Agent process for concurrent access and updates.
The state is managed by the `Phoenix.SocketClient.State` struct.
"""
use Agent
alias Phoenix.SocketClient.Message
alias Phoenix.SocketClient.State
alias Phoenix.SocketClient.Telemetry
@heartbeat_interval 30_000
@reconnect_interval 60_000
@default_transport Phoenix.SocketClient.Transports.Websocket
@doc """
Starts the Agent with the given configuration options.
## Parameters
* `opts` - Keyword list or map of configuration options
## Examples
{:ok, pid} = Phoenix.SocketClient.Agent.start_link(url: "ws://localhost:4000/socket")
"""
@spec start_link(keyword() | map()) :: {:ok, pid()} | {:error, term()}
def start_link(opts) do
opts = if Keyword.keyword?(opts), do: opts, else: Map.to_list(opts)
Agent.start_link(fn -> init_state(opts) end)
end
@doc """
Retrieves a value from the state by key.
## Parameters
* `pid` - The Agent PID
* `key` - The key to retrieve
## Examples
value = Phoenix.SocketClient.Agent.get(pid, :url)
"""
@spec get(pid(), atom() | String.t()) :: any()
def get(pid, key) do
Agent.get(pid, fn state ->
Map.get(state, key, Map.get(state.custom, key))
end)
end
@doc """
Retrieves the entire state map.
"""
@spec get_state(pid()) :: map()
def get_state(pid) do
Agent.get(pid, & &1)
end
@doc """
Updates the state with a new key-value pair.
## Parameters
* `pid` - The Agent PID
* `key` - The key to set
* `value` - The value to associate with the key
## Examples
:ok = Phoenix.SocketClient.Agent.put(pid, :status, :connected)
"""
@spec put(pid(), atom() | String.t(), any()) :: :ok
def put(pid, key, value) do
Agent.update(pid, fn state ->
if Map.has_key?(state, key) do
Map.put(state, key, value)
else
custom = Map.put(state.custom, key, value)
%State{state | custom: custom}
end
end)
end
@doc """
Checks if the socket is connected.
"""
@spec connected?(pid()) :: boolean()
def connected?(pid) do
get(pid, :status) == :connected
end
@doc """
Pops all messages pending to be sent.
"""
@spec pop_all_to_send(pid()) :: [Message.t()]
def pop_all_to_send(pid) do
Agent.get_and_update(pid, fn state ->
to_send = state.to_send_r |> Enum.reverse()
{to_send, %State{state | to_send_r: []}}
end)
end
@doc """
Updates the status of a channel.
"""
@spec update_channel_status(pid(), pid(), String.t(), atom(), map() | nil) :: :ok
def update_channel_status(pid, channel_pid, topic, status, params \\ nil) do
Agent.update(pid, fn state ->
old_channel_data = Map.get(state.joined_channels, topic, %{})
old_status = Map.get(old_channel_data, :status)
new_channel_data =
if params do
Map.put(old_channel_data, :params, params)
else
old_channel_data
end
|> Map.put(:status, status)
|> Map.put(:pid, channel_pid)
joined_channels = Map.put(state.joined_channels, topic, new_channel_data)
if old_status != status do
Telemetry.channel_status_changed(channel_pid, topic, old_status, status)
end
%State{state | joined_channels: joined_channels}
end)
end
@doc """
Removes a channel from the state.
"""
@spec remove_channel(pid(), String.t()) :: :ok
def remove_channel(pid, topic) do
Agent.update(pid, fn state ->
joined_channels = Map.delete(state.joined_channels, topic)
%State{state | joined_channels: joined_channels}
end)
end
defp init_state(opts) do
defaults = %{
json_library: Jason,
reconnect: true,
auto_connect: true,
vsn: "2.0.0",
url: "ws://localhost:4000/socket/websocket",
params: %{},
headers: [],
heartbeat_interval: @heartbeat_interval,
reconnect_interval: @reconnect_interval,
transport: @default_transport,
transport_opts: [],
sup_pid: nil,
reconnect_timer: nil,
status: :disconnected,
transport_pid: nil,
to_send_r: [],
ref: 0,
custom: %{},
registry_name: Registry.Channel,
join_channels: [],
default_channel_module: Phoenix.SocketClient.Channel.EchoRoom,
default_channel_params: %{}
}
config = Map.merge(defaults, Enum.into(opts, %{}))
custom_opts = Map.drop(config, Map.keys(defaults))
config = Map.put(config, :custom, custom_opts)
url = config.url
uri = URI.parse(url)
query_params = Map.merge(%{"vsn" => config.vsn}, config.params)
query = URI.encode_query(query_params)
base_url =
if uri.query do
url_parts = String.split(url, "?", parts: 2)
base = Enum.at(url_parts, 0)
existing_query = Enum.at(url_parts, 1)
if existing_query && String.trim(existing_query) != "" do
base <> "?" <> existing_query <> "&" <> query
else
base <> "?" <> query
end
else
url <> "?" <> query
end
transport_opts =
config.transport_opts
|> Keyword.put_new(:extra_headers, config.headers)
|> Keyword.put_new(:keepalive, config.heartbeat_interval)
%State{
url: base_url,
json_library: config.json_library,
params: config.params,
vsn: config.vsn,
auto_connect: config.auto_connect,
reconnect: config.reconnect,
reconnect_interval: config.reconnect_interval,
reconnect_timer: config.reconnect_timer,
status: config.status,
serializer: Message.serializer(config.vsn),
transport: config.transport,
transport_opts: transport_opts,
transport_pid: config.transport_pid,
to_send_r: config.to_send_r,
ref: config.ref,
sup_pid: config.sup_pid,
headers: config.headers,
custom: config.custom,
registry_name: config.registry_name,
join_channels: config.join_channels,
default_channel_module: config.default_channel_module,
default_channel_params: config.default_channel_params
}
end
end