ChannelServer

ChannelServer module: runtime helpers for Phoenix-style channel processes.

This module provides the building blocks for implementing channel processes using March's actor system. The actual actor declarations must be provided by the application (actors cannot be spawned from closures in March).

A channel process is an actor that:

Receives a ChannelJoin message with topic/params.
Calls the handler's join/3 callback.
On success, subscribes to PubSub for the topic.
Loops receiving ChannelMailbox messages and dispatching to callbacks.
On ChannelLeave, unsubscribes from PubSub and calls leave/3.

Example actor using these helpers (sketch: cfg is a ChannelConfig built with make_config, handle_in the application's callback; the module has no init/handle_in/leave wrappers of its own):

actor MyChannelProcess do
  state = ...

  on ChannelIn(ref, event, payload) ->
    ChannelServer.apply_result(cfg, handle_in(ref, event, payload, state), ref)

  on ChannelBroadcast(topic, event, payload) ->
    ChannelServer.handle_broadcast(cfg, event, payload)
    state

  on ChannelLeave(reason) ->
    ChannelServer.do_leave(cfg, state, reason)
    state
end

PubSub shard actors are similarly declared by the framework user:

actor PubSubShard do
  state = PubSub.new_state()

  on PubSubSubscribe(topic, pid) ->
    PubSub.subscribe_state(state, topic, pid)

  on PubSubUnsubscribe(topic, pid) ->
    PubSub.unsubscribe_state(state, topic, pid)

  on PubSubBroadcast(topic, excl, event, payload) ->
    PubSub.broadcast_to(PubSub.get_subscribers(state, topic), excl, event, payload)
    state
end

Types

typeJoinResultJoinResult(a) = JoinOk(a) | JoinErr(String)#

Result of a channel join attempt. JoinOk(state) — join accepted; state is the handler's initial state JoinErr(reason) — join rejected with a reason string

typeChannelConfigChannelConfig(a, b, c, d, e) = ChannelConfig(#

Runtime configuration for a single channel process. socket — the Channel.Socket for this join join_fn — fn(topic, params, socket) -> Result(state, String) handle_in_fn — fn(ref, event, payload, state, socket) -> HandleResult handle_info_fn — fn(msg, state, socket) -> HandleResult leave_fn — fn(reason, state, socket) -> Unit pubsub_shard — pid of the PubSub shard actor for this topic

Functions

fnapply_resultapply_result(cfg, result, ref)#

Apply a HandleResult: if Reply, send the reply to the transport; return the new state. For Stop, call do_leave. Returns the new channel state. A Reply(status, payload, st) is pushed to the transport as ChannelPush("phx_reply", serialize(reply_msg(join_ref, ref, topic, status, payload))). Stop runs do_leave but sends the client no phx_close or phx_error. Must be called from within an actor handler (requires actor runtime).

let state2 = ChannelServer.apply_result(cfg, Channel.reply("ok", "{}", state), ref)

fnconfig_shardconfig_shard(cfg)#

The PubSub shard pid of a config.

Examples

march> let cfg = ChannelServer.make_config(Channel.new_socket("room:lobby", 10, 11, "0"), fn (_t, _p, _s) -> Ok(0), fn (_r, _e, _p, st, _s) -> Channel.noreply(st), fn (_m, st, _s) -> Channel.noreply(st), fn (_r, _st, _s) -> (), 99)
march> ChannelServer.config_shard(cfg)
99

fnconfig_socketconfig_socket(cfg)#

The Channel.Socket of a config.

Examples

march> let cfg = ChannelServer.make_config(Channel.new_socket("room:lobby", 10, 11, "0"), fn (_t, _p, _s) -> Ok(0), fn (_r, _e, _p, st, _s) -> Channel.noreply(st), fn (_m, st, _s) -> Channel.noreply(st), fn (_r, _st, _s) -> (), 99)
march> Channel.socket_topic(ChannelServer.config_socket(cfg))
"room:lobby"

fndo_joindo_join(cfg, topic, params_json)#

Perform the join handshake. Calls join_fn(topic, params_json, socket); on Ok(state) subscribes the calling process (self()) to topic in the config's shard and returns JoinOk(state), on Err(reason) returns JoinErr(reason) without subscribing. Sends no reply to the client. Must be called from within an actor handler (requires actor runtime).

match ChannelServer.do_join(cfg, "room:lobby", "{}") do
  ChannelServer.JoinOk(st) -> st
  ChannelServer.JoinErr(why) -> ...
end

fndo_leavedo_leave(cfg, st, reason_str)#

Perform the leave teardown. Unsubscribes the calling process from the socket's topic in the config's shard, then returns leave_fn(reason_str, st, socket). Must be called from within an actor handler (requires actor runtime).

ChannelServer.do_leave(cfg, state, "user_left")

fnhandle_broadcasthandle_broadcast(cfg, event, payload_json)#

Handle a ChannelBroadcast message received by a channel process. Forwards the event+payload to the transport pid as ChannelPush. Must be called from within an actor handler (requires actor runtime).

ChannelServer.handle_broadcast(cfg, "new_msg", payload)

fnhandle_broadcast_fromhandle_broadcast_from(cfg, excl_pid, event, payload_json)#

Handle a ChannelBroadcastFrom message. Forwards event+payload to the transport unless this process (self()) is excl_pid. Must be called from within an actor handler (requires actor runtime).

ChannelServer.handle_broadcast_from(cfg, sender_pid, "typing", "{}")

fnmake_configmake_config(socket, join_fn, handle_in_fn, handle_info_fn, leave_fn, shard_pid)#

A ChannelConfig from the socket, the four callbacks (see the type's comment for their shapes) and the PubSub shard pid for the topic.

Examples

march> let cfg = ChannelServer.make_config(Channel.new_socket("room:lobby", 10, 11, "0"), fn (_t, _p, _s) -> Ok(0), fn (_r, _e, _p, st, _s) -> Channel.noreply(st), fn (_m, st, _s) -> Channel.noreply(st), fn (_r, _st, _s) -> (), 99)
march> ChannelServer.config_shard(cfg)
99

fnselect_shardselect_shard(shards, topic)#

Select the shard pid for topic from a list of shard pids. Uses PubSub.topic_shard for consistent routing (the same topic always picks the same index for the same list length on the same platform). Panics (division by zero) if shards is empty.

Examples

march> ChannelServer.select_shard([42], "room:lobby")
42

fnshard_handle_broadcastshard_handle_broadcast(ps_state, topic, excluded_opt, event, payload_json)#

Process a PubSubBroadcast message inside a shard actor. Fans out to all subscribers, optionally excluding a pid. Returns the PubSubState unchanged. Requires actor runtime (actor_cast).

ChannelServer.shard_handle_broadcast(st, "room:lobby", None, "new_msg", "{}")

fnshard_handle_subscribeshard_handle_subscribe(ps_state, topic, pid)#

Process a PubSubSubscribe message inside a shard actor. Returns the new PubSubState (PubSub.subscribe_state).

let st2 = ChannelServer.shard_handle_subscribe(st, "room:lobby", pid)

fnshard_handle_unsubscribeshard_handle_unsubscribe(ps_state, topic, pid)#

Process a PubSubUnsubscribe message inside a shard actor. Returns the new PubSubState (PubSub.unsubscribe_state).

Examples

march> PubSub.subscriber_count(ChannelServer.shard_handle_unsubscribe(PubSub.new_state(), "room:lobby", 7), "room:lobby")
0