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:

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

The socket will automatically attempt to connect when it starts. If the socket becomes disconnected, it will attempt to reconnect automatically, and by default every joined channel is re-joined on the new connection (see Reconnections). Joining blocks until the connection is up, so you do not need to poll ChannelClient.Socket.connected?/1 first.

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)

push/4 returns {:ok, response} when the server replies with an ok status, and {:error, reason} for everything else (server error replies, timeouts, unencodable payloads). It never raises or exits.

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".

Event handlers

If you prefer callbacks over mailbox pattern matching, pass a handler/1 fun to join. It receives every %Message{} for the topic and runs inside the channel process (crashes in it are logged, never propagated):

handler = fn %ChannelClient.Message{event: event, payload: payload} ->
  IO.puts("#{event}: #{inspect(payload)}")
end

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

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 using ChannelClient.Plug (write just call/2, or add your own init/1) 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, the socket reconnects automatically and re-joins every topic it was subscribed to (rejoin?: true, the default). Delivery simply resumes — callers need no re-join logic. Failed rejoins are reported to the channel's owner as a %ChannelClient.Message{} with event "phx_error".

With rejoin?: false, channels are unregistered on disconnect instead and each joining process receives a %ChannelClient.Message{} with 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.

While connected, the socket sends Phoenix protocol heartbeats every heartbeat_interval ms (default 30_000) so servers keep the connection open; set it to :infinity to disable.

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. Pass :name (and optionally :id) inline so it can be referenced from your channel:

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

  children = [
    {ChannelClient.Socket, Keyword.merge(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, Keyword.merge(socket_1_opts, name: :socket_1)},
    {ChannelClient.Socket, Keyword.merge(socket_2_opts, name: :socket_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.