ChannelClient.Plug behaviour (channel_client v0.1.0)

Copy Markdown

A behaviour for pluggable message middleware, modelled after Phoenix's Plug.

Plugs run in two pipelines owned by ChannelClient.Socket:

  • :outbound_plugs - run on every frame the client is about to send (joins, leaves and pushes), before ref assignment and encoding.
  • :inbound_plugs - run on every decoded frame arriving from the server, before it is routed to a channel.

A plug receives the %ChannelClient.Message{} and its initialized opts, and returns one of:

  • {:cont, message} - continue the pipeline with (optionally transformed) message.
  • {:halt, reason} - stop the pipeline. Outbound frames are not sent; inbound frames are dropped. Synchronous callers receive {:error, {:halted, reason}}.

Note that the socket assigns ref and join_ref after the outbound pipeline runs, so plugs may freely transform payload, topic or event; those two fields are always protocol-correct on the wire.

Example

defmodule MyApp.InjectTenant do
  @behaviour ChannelClient.Plug

  def init(opts), do: opts

  def call(message, tenant_id) do
    {:cont, update_in(message.payload, &Map.put(&1, "tenant_id", tenant_id))}
  end
end

# In your socket options:
outbound_plugs: [{MyApp.InjectTenant, "acme"}]

Plug specs

Both module plugs and anonymous function plugs are supported:

  • MyPlug - equivalent to {MyPlug, []}
  • {MyPlug, opts} - opts are passed through MyPlug.init/1
  • fun where fun/2 - equivalent to {fun, []}
  • {fun, opts} - opts passed as-is to the function

Invalid specs raise ArgumentError when the socket starts. Exceptions raised inside a plug are caught, logged and treated as a halt, so a faulty plug cannot crash the socket.

Summary

Callbacks

Runs the plug on a message. Invoked for every frame flowing through the attached pipeline.

Prepares or validates a plug's options. Invoked once at socket startup.

Functions

Compiles a list of plug specs into runnable form.

Runs a compiled pipeline over message.

Types

compiled()

@type compiled() ::
  {module() | (ChannelClient.Message.t(), opts() -> result()), opts()}

opts()

@type opts() :: term()

result()

@type result() :: {:cont, ChannelClient.Message.t()} | {:halt, term()}

spec()

@type spec() ::
  module()
  | {module(), opts()}
  | (ChannelClient.Message.t(), opts() -> result())
  | {(ChannelClient.Message.t(), opts() -> result()), opts()}

Callbacks

call(t, opts)

@callback call(ChannelClient.Message.t(), opts()) :: result()

Runs the plug on a message. Invoked for every frame flowing through the attached pipeline.

init(opts)

(optional)
@callback init(opts()) :: opts()

Prepares or validates a plug's options. Invoked once at socket startup.

Optional; defaults to returning the opts unchanged.

Functions

compile(specs)

@spec compile([spec()]) :: [compiled()]

Compiles a list of plug specs into runnable form.

Raises ArgumentError on invalid specs so misconfiguration surfaces at socket startup rather than at runtime.

run(plugs, message)

@spec run([compiled()], ChannelClient.Message.t()) :: result()

Runs a compiled pipeline over message.

Returns {:cont, message} after all plugs ran, or the first {:halt, reason}. An exception inside a plug halts with {:plug_raised, exception} as the reason.

Implemented with explicit recursion rather than Enum.reduce_while, whose {:cont, acc} / {:halt, acc} accumulator protocol would unwrap our own identically-shaped results.