ChannelSocket

ChannelSocket module: WebSocket transport adapter for Channels.

ChannelSocket integrates the Channel wire protocol with March's WebSocket server. A socket endpoint is mounted in the HTTP router and handles:

WebSocket upgrade for incoming connections.
Authentication via the user-supplied connect/2 callback.
Routing incoming phx_join messages to the correct channel handler.
Forwarding ChannelPush messages from channel processes to the wire.

Usage:

fn my_router() do
  HttpServer.new(4000)
  |> HttpServer.plug(ChannelSocket.plug_for(ChannelSocket.make_config("/socket", my_routes, connect_fn)))
  |> HttpServer.listen()
end

-- NOTE: connect_fn is stored in the config but nothing in this module
-- calls it yet; connections are not authenticated.
fn connect_fn(params, socket) do
  match verify_token(params) do
  | Ok(user) -> Ok(Channel.assign(socket, "user", user))
  | Err(_)   -> Err("unauthorized")
  end
end

Socket route list: Each ChannelRoute maps a topic pattern to a handler name string. On join, the handler name is used to look up the handler functions in the environment (see ChannelServer for the handler function protocol).

Transport loop (per client connection): The WebSocket handler runs a loop reading frames from the wire, deserialising them as ChannelMsg, and dispatching to channel processes. Outbound ChannelPush messages from channel processes are serialised and sent back over the WebSocket.

See also: channel.march, channel_server.march, websocket.march.

Types

typeSocketConfigSocketConfig(a) = SocketConfig(#

Configuration for a socket endpoint. path — URL path prefix, e.g. "/socket" routes — list of ChannelRoute (topic pattern → handler name) connect_fn — fn(params_map, socket) -> Result(Socket, String) shards — list of PubSub shard pids (populated at server start)

typeActiveChannelsActiveChannels = ActiveChannels(Map(String, Int))#

Active channels for one socket connection: Map(topic, channel_pid). Used by the socket loop to route inbound messages to the right channel process and to send leave notifications on disconnect.

Functions

fnall_channelsall_channels(ac)#

All (topic, pid) pairs in the registry, in the map's hash-traversal order.

fnconfig_pathconfig_path(cfg)#

The URL path prefix of a SocketConfig.

ChannelSocket.config_path(ChannelSocket.make_config("/socket", [], connect_fn))   -- "/socket"

fnconfig_routesconfig_routes(cfg)#

The ChannelRoute list of a SocketConfig, in the order given.

fnconfig_shardsconfig_shards(cfg)#

The PubSub shard pids of a SocketConfig ([] from make_config).

fnderegister_channelderegister_channel(ac, topic)#

Remove topic from the registry (after leave). A topic that is not registered leaves it unchanged.

ChannelSocket.lookup_channel(ChannelSocket.deregister_channel(ac, "room:1"), "room:1")   -- None

fnempty_channelsempty_channels()#

An empty active-channels registry.

ChannelSocket.all_channels(ChannelSocket.empty_channels())   -- []

fnfind_routefind_route(routes, topic)#

Find the first ChannelRoute whose pattern matches topic (PubSub.topic_matches), so list more specific routes first. Returns Some(route) if found, None otherwise.

let routes = [Channel.channel_route("room:lobby", "Lobby"), Channel.channel_route("room:*", "Room")]
Option.map(ChannelSocket.find_route(routes, "room:42"), Channel.route_handler)   -- Some("Room")
ChannelSocket.find_route(routes, "game:1")                                       -- None

fnhas_routehas_route(routes, topic)#

True when any route in routes matches topic.

ChannelSocket.has_route([Channel.channel_route("room:*", "Room")], "room:1")   -- true

fnlookup_channellookup_channel(ac, topic)#

The channel process pid registered for topic, or None when the topic is not joined on this connection.

ChannelSocket.lookup_channel(ChannelSocket.empty_channels(), "room:1")   -- None

fnmake_configmake_config(path, routes, connect_fn)#

A SocketConfig for the endpoint at path with routes and connect_fn, and no PubSub shards yet.

let cfg = ChannelSocket.make_config("/socket", [Channel.channel_route("room:*", "RoomChannel")], connect_fn)
ChannelSocket.config_shards(cfg)   -- []

fnmake_config_with_shardsmake_config_with_shards(path, routes, connect_fn, shards)#

A SocketConfig like make_config, with the pids of already-started PubSub shard actors.

let cfg = ChannelSocket.make_config_with_shards("/socket", routes, connect_fn, [shard1, shard2])

fnparse_frameparse_frame(frame_text)#

Parse a WebSocket TextFrame string as a ChannelMsg. Returns Some(msg) or None on parse failure. Same as Channel.parse, with its limits (no spaces after colons, no null fields, payload last).

ChannelSocket.parse_frame("{\"join_ref\":\"0\",\"ref\":\"1\",\"topic\":\"room:1\",\"event\":\"phx_join\",\"payload\":{}}")
-- Some(ChannelMsg("0", "1", "room:1", "phx_join", "{}"))

fnplug_forplug_for(cfg)#

Build a Conn → Conn plug function that handles WebSocket upgrade for the socket path. Mount with HttpServer.plug/2.

The plug:

Checks that the request path starts with socket_path (a plain

   string prefix: "/socket" also matches "/socketfoo"); any other
   request passes through unchanged.
2. Upgrades to WebSocket.
3. In the WS handler loop: reads frames, parses them as ChannelMsgs,
   and routes them; unparseable frames are ignored, and on Close every
   active channel is sent `ChannelLeave("disconnect")`.

Usage: HttpServer.new(4000) |> HttpServer.plug(ChannelSocket.plug_for(cfg)) |> HttpServer.listen()

fnregister_channelregister_channel(ac, topic, pid)#

Record pid as the channel process for topic, replacing any pid already registered for it.

let ac = ChannelSocket.register_channel(ChannelSocket.empty_channels(), "room:1", 42)
ChannelSocket.lookup_channel(ac, "room:1")   -- Some(42)

fnroute_msgroute_msg(ac, msg, routes, cfg, transport_pid)#

Route a parsed ChannelMsg to the appropriate channel process or handle it as a socket-level message (phx_join, phx_leave, phx_heartbeat). Returns an updated ActiveChannels or triggers side effects. By event:

phx_join: with no matching route, pushes a phx_error ("unmatched

topic") to `transport_pid`; with one, registers `transport_pid` itself as
the topic's channel pid. No channel process is spawned and no join reply
is sent yet.

phx_leave: casts ChannelLeave("user_left") to the topic's channel and

deregisters it (no-op when not joined).

phx_heartbeat: pushes a phx_reply with status "ok" on topic "phoenix".
anything else: casts ChannelIn(ref, event, payload) to the topic's

channel, or pushes a phx_error ("not joined") when there is none.

routes is used; cfg is not. Requires actor runtime.

let active2 = ChannelSocket.route_msg(active, msg, routes, cfg, transport_pid)

fnserialize_msgserialize_msg(msg)#

Serialize a ChannelMsg for sending as a WebSocket TextFrame. Same as Channel.serialize.

ChannelSocket.serialize_msg(Channel.close_msg("0", "room:1"))
-- "{\"join_ref\":\"0\",\"ref\":\"\",\"topic\":\"room:1\",\"event\":\"phx_close\",\"payload\":{}}"