Plugs: middleware for channel messages

ChannelClient supports a pluggable pipeline architecture modelled after Phoenix's Plug. Plugs are small, composable modules (or functions) that inspect, transform, or block messages as they flow through the socket.

The two pipelines

A ChannelClient.Socket runs two independent pipelines:

• outbound_plugs run on every frame the client is about to send: joins, leaves and pushes. They run before the socket assigns ref / join_ref stamps and encodes the frame, so the wire format is always protocol-correct no matter what a plug does to the payload. A halt stops the frame from ever being sent: synchronous callers get {:error, {:halted, reason}}, async pushes are logged and dropped.

• inbound_plugs run on every decoded server frame before it is routed to the matching channel. Halting a message drops it silently — the joining process never sees it.

{:ok, socket} =
  ChannelClient.Socket.start_link(
    url: "ws://localhost:4000/socket/websocket",
    inbound_plugs: [
      {ChannelClient.Plugs.FilterEvents, events: ["presence_diff"]}
    ],
    outbound_plugs: [
      MyApp.InjectTenant
    ]
  )

Writing a plug

The quickest way is use ChannelClient.Plug, which injects the behaviour and a default init/1 (opts pass through unchanged) — you only write call/2:

defmodule MyApp.InjectTenant do
  use ChannelClient.Plug

  @impl true
  def call(message, tenant_id) do
    {:cont, %{message | payload: Map.put(payload_or_empty(message.payload), "tenant_id", tenant_id)}}
  end

  defp payload_or_empty(nil), do: %{}
  defp payload_or_empty(payload) when is_map(payload), do: payload
  defp payload_or_empty(other), do: other
end

To validate or preprocess options once at startup, define your own init/1 (it overrides the injected default):

@impl true
def init(opts) do
  Keyword.validate!(opts, [:tenant_id])
end

Prefer explicitness over macros? The plain behaviour works identically:

defmodule MyApp.InjectTenant do
  @behaviour ChannelClient.Plug

  @impl true
  def init(opts), do: opts

  @impl true
  def call(message, opts), do: {:cont, ...}
end

Plugs run in the order they are listed in the pipeline option.

The call/2 callback returns either:

• {:cont, message} — pass the (optionally transformed) message on;
• {:halt, reason} — stop the pipeline for this message.

A plug that raises or exits is caught, logged and treated as a halt ({:halt, {:plug_raised, ...}}) — a faulty plug can never crash the socket. Returning anything but {:cont, _} / {:halt, _} halts with {:bad_plug_result, value}.

Plug specs

Every slot in a pipeline accepts any of these forms:

Invalid specs raise ArgumentError when the socket starts, so misconfiguration fails fast. Exceptions raised inside a plug are caught, logged, and treated as a halt — a faulty plug can never crash the socket.

Built-in plugs

ChannelClient.Plugs.Logger

Logs each message's topic and event as it crosses the pipeline.

inbound_plugs: [{ChannelClient.Plugs.Logger, level: :info}]

Options: :level (default :debug), :label.

ChannelClient.Plugs.FilterEvents

Halts messages whose event does not match a configured list — a blocklist via :events, or an allowlist via :only:

# keep noisy broadcasts away from your processes:
inbound_plugs: [{ChannelClient.Plugs.FilterEvents, events: ["phx_reply", "presence_diff"]}]

# or subscribe to just the events you care about:
inbound_plugs: [{ChannelClient.Plugs.FilterEvents, only: ["user:entered"]}]

Options: :events (blocklist) or :only (allowlist) — give one, not both — and :reason (halt reason, default :filtered). A single event name is accepted wherever a list is.

Function plugs

For quick one-offs, anonymous functions keep everything inline:

outbound_plugs: [
  fn msg, _opts ->
    {:cont, %{msg | payload: Map.put(msg.payload || %{}, "sent_at", DateTime.utc_now())}}
  end
]

Testing plugs

Because plugs are plain functions over %ChannelClient.Message{}, they unit test without a socket:

test "injects the tenant" do
  message = %ChannelClient.Message{topic: "rooms:lobby", event: "new:msg", payload: %{}}
  assert {:cont, %{payload: %{"tenant_id" => "acme"}}} = MyApp.InjectTenant.call(message, "acme")
end

See test/channel_client/socket_protocol_test.exs in this repository for examples of driving whole pipelines against a stub transport.