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

Implement the ChannelClient.Plug behaviour:

defmodule MyApp.InjectTenant do
  @behaviour ChannelClient.Plug

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

  @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

The call/2 callback returns either:

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

init/1 is optional; use it to validate options once at startup instead of per message.

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 appears in a blocklist — useful for keeping noisy broadcasts away from your processes:

inbound_plugs: [{ChannelClient.Plugs.FilterEvents, events: ["phx_reply", "presence_diff"]}]

Options: :events (required list), :reason (halt reason, default :filtered).

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.