ChannelClient

Channel client for connecting to Phoenix Channels from Elixir.

Rework from phoenix_client library.

Installation

Add channel_client and a json library as dependencies in your mix.exs file. jason is specified as the default json library.

def deps do
  [
    {:channel_client, "~> 0.12"},
    {:jason, "~> 1.4"}
  ]
end

If you choose to use a different json library, you can set it through the socket options with the :json_library key (or its alias, :serializer).

Usage

There are two things required to connect to a phoenix server using channels, a ChannelClient.Socket and a ChannelClient.Channel. The socket establishes the connection to the remote socket. The channel takes a topic and is used to join a remote channel. In the following example we will assume that we are attempting to communicate with a locally running phoenix server with a RoomChannel with the topic room:lobby configured to route to the RoomChannel in the UserSocket.

First, Lets create a client socket:

socket_opts = [
  url: "ws://localhost:4000/socket/websocket"
]

{:ok, socket} = ChannelClient.Socket.start_link(socket_opts)

The socket will automatically attempt to connect when it starts. If the socket becomes disconnected, it will attempt to reconnect automatically. Please note that start_link is not synchronous so you must wait for the socket to become connected before attempting to join a channel.

You can control how frequently the socket will attempt to reconnect by setting reconnect_interval in the socket_opts.

Next, we will create a client channel and join the remote.

{:ok, _response, channel} = ChannelClient.Channel.join(socket, "rooms:lobby")

Now that we have successfully joined the channel, we are ready to push and receive new messages. Pushing a message can be done synchronously or asynchronously. If you require a reply, or want to institute a time out, you can call push. If you do not require a response, you can call push_async.

In this example, we will assume the server channel has the following handle_in callbacks:

  def handle_in("new:msg", message, socket) do
    {:reply, {:ok, message}, socket}
  end

  def handle_in("new:msg_async", _message, socket) do
    {:noreply, socket}
  end

message = %{hello: :world}
{:ok, ^message} = ChannelClient.Channel.push(channel, "new:msg", message)
:ok = ChannelClient.Channel.push_async(channel, "new:msg_async", message)

Payloads may be any value the configured JSON library can encode. Payloads that cannot be encoded return {:error, reason} from push/4, and are logged and dropped by push_async/3 — the socket connection stays healthy either way.

Messages that are pushed or broadcasted to the client channel will be sent to the pid that called join. Messages will be of the of the struct %ChannelClient.Message{}.

In this example we will assume the server channel has the following handle_in callback

  def handle_in("new:msg", message, socket) do
    push(socket, "incoming:msg", message)
    {:reply, :ok, socket}
  end

message = %{hello: :world}
{:ok, ^message} = ChannelClient.Channel.push(channel, "new:msg", message)
flush

%ChannelClient.Message{
  channel_pid: #PID<0.186.0>,
  event: "incoming:msg",
  payload: %{"hello" => "world"},
  ref: nil,
  topic: "room:lobby"
}

Replies that arrive without a matching synchronous push (for example replies to push_async) are forwarded to the joining process as %ChannelClient.Message{} structs with the event "phx_reply".

Wire formats

JSON is the default, but the format layer is pluggable. ETF (Erlang terms), TOON and BTOON ship built in:

{:ok, socket} =
  ChannelClient.Socket.start_link(url: "ws://localhost:4000/socket/websocket", format: :etf)

# TOON/BTOON delegate to {:toon_ex, "~> 1.5"} (optional dependency):
{:ok, socket} = ChannelClient.Socket.start_link(url: "...", format: :toon)
{:ok, socket} = ChannelClient.Socket.start_link(url: "...", format: :btoon)

Custom formats (MessagePack, protobufs, ...) implement the ChannelClient.Format behaviour and plug in with format: MyApp.MsgPack. Legacy options (:vsn, :json_library, :serializer) keep working. See the Formats guide.

Pluggable architecture

Sockets support Plug-style middleware for messages, in two pipelines:

{:ok, socket} =
  ChannelClient.Socket.start_link(
    url: "ws://localhost:4000/socket/websocket",
    inbound_plugs: [
      {ChannelClient.Plugs.FilterEvents, events: ["presence_diff"]}
    ],
    outbound_plugs: [
      fn msg, _opts -> {:cont, %{msg | payload: Map.put(msg.payload || %{}, "sent_at", DateTime.utc_now())}} end
    ]
  )

A plug is a module implementing the ChannelClient.Plug behaviour (init/1 + call/2) or a plain fun/2. It returns {:cont, message} to pass the message along (optionally transformed) or {:halt, reason} to block it — halted outbound frames surface as errors to sync callers, halted inbound frames never reach your processes. Faulty plugs are logged and treated as halts; they cannot crash the socket.

Built-ins ship under ChannelClient.Plugs (Logger, FilterEvents). See the Plugs guide for the full walkthrough.

Telemetry & tracing

Every connection, message and channel operation emits standard :telemetry events, including start/stop/exception span triples you can feed straight into dashboards or tracing backends:

:telemetry.attach(
  "my-handler",
  [:channel_client, :push, :stop],
  fn _name, %{duration: d}, meta, _ ->
    Logger.debug("push #{meta.event} -> #{meta.result}")
  end,
  nil
)

Payload contents are never included in event metadata. See the Telemetry guide for the full event catalog.

Reconnections

When the underlying connection drops, all channels are unregistered and each joining process receives a %ChannelClient.Message{} with the event "phx_close" (clean close) or "phx_error". Callers are expected to react by joining again once the socket reports connected through ChannelClient.Socket.connected?/1.

Both :text and :binary WebSocket frames are supported for inbound messages.

Common configuration

You can configure the socket to be started in your main application supervisor. You will need to name the socket so it can be referenced from your channel.

  socket_opts =
    Application.get_env(:channel_client, :socket)

  children = [
    {ChannelClient.Socket, {socket_opts, name: ChannelClient.Socket}}
  ]

You will need a socket for each server you are connecting to. Here is an example for connecting to multiple remote servers.

  socket_1_opts =
    Application.get_env(:channel_client, :socket_1)
  socket_2_opts =
    Application.get_env(:channel_client, :socket_2)

  children = [
    {ChannelClient.Socket, {socket_1_opts, name: :socket_1, id: :socket_id_1}},
    {ChannelClient.Socket, {socket_2_opts, name: :socket_2, id: :socket_id_2}}
  ]

Channels are usually constructed in a process such as a GenServer. Here is an example of how this is typically used.

defmodule MyApp.Worker do
  use GenServer

  alias ChannelClient.{Socket, Channel, Message}

  # start_link ...

  def init(_opts) do
    {:ok, _response, channel} = Channel.join(Socket, "room:lobby")
    {:ok, %{
      channel: channel
    }}
  end

  # do some work, call `Channel.push` ...

  def handle_info(%Message{event: "incoming:msg", payload: payload}, state) do
    IO.puts "Incoming Message: #{inspect payload}"
    {:noreply, state}
  end
end

License

Apache-2.0. Based on the original phoenix_client work.