Layr8.Channel (layr8 v0.2.10)

Copy Markdown View Source

Phoenix Channel transport over WebSocket.

Implements the Phoenix Channel V2 wire protocol:

[join_ref, ref, topic, event, payload]

Connection Flow

  1. Connects to {node_url}?api_key={key}&vsn=2.0.0
  2. Sends phx_join with registered protocols and DID spec
  3. Fires heartbeat every 30 seconds
  4. Auto-reconnects with exponential backoff (1s → 30s) on disconnect

One socket, many DIDs

The DID passed to start_link/1 is the primary one: its topic (plugins:<did>) is joined by connect/2 and is what every …_on/3,4 function addresses as :primary. join_did/5 joins additional plugins:<did> topics on the same WebSocket — cloud-node's plugin socket matches channel("plugins:*", …), so a socket may hold any number of them.

Additional DIDs are addressed as {:did, did} and are re-joined automatically after a reconnect: losing them on a dropped socket would be a silent failure — messages for those DIDs simply stop arriving, with nothing to observe on this side.

Callbacks

  • on_message/1 — called with the decoded payload of inbound "message" events on the primary topic; each additional DID carries its own on_message given to join_did/5
  • on_disconnect/1 — called with the error reason on disconnect
  • on_reconnect/0 — called after successful reconnection

Not meant to be used directly; interact through Layr8.Client.

Summary

Types

Which joined topic a call addresses: the DID given to start_link/1 (:primary) or one joined later with join_did/5.

Functions

Returns the DID assigned by the cloud-node (for ephemeral identities).

Returns the capabilities advertised by the cloud-node on join.

Returns a specification to start this module under a supervisor.

Closes the WebSocket connection.

Connects to the cloud-node WebSocket and joins the Phoenix channel.

Joins an additional plugins:<did> topic on the already-open WebSocket.

Returns the additional DIDs currently joined (excludes the primary DID).

Leaves an additional DID's topic (phx_leave) and forgets it, so a later reconnect does not bring it back. The socket and every other topic stay up.

Sends an acknowledgment for a list of message IDs.

send_ack/2 on a specific joined topic — see target/0.

Sends a message event fire-and-forget (no reply tracking).

Sends a message event and waits for a server acknowledgment.

send_msg/3 on a specific joined topic — see target/0.

Starts the Channel GenServer.

Types

callbacks()

@type callbacks() :: %{
  on_message: on_message_fn(),
  on_disconnect: on_disconnect_fn() | nil,
  on_reconnect: on_reconnect_fn() | nil
}

on_disconnect_fn()

@type on_disconnect_fn() :: (term() -> any())

on_message_fn()

@type on_message_fn() :: (map() -> any())

on_reconnect_fn()

@type on_reconnect_fn() :: (-> any())

start_opts()

@type start_opts() :: %{
  :ws_url => String.t(),
  :api_key => String.t(),
  :agent_did => String.t(),
  :callbacks => callbacks(),
  optional(:did_spec) => map()
}

target()

@type target() :: :primary | {:did, String.t()}

Which joined topic a call addresses: the DID given to start_link/1 (:primary) or one joined later with join_did/5.

Functions

assigned_did(pid)

@spec assigned_did(pid()) :: String.t()

Returns the DID assigned by the cloud-node (for ephemeral identities).

capabilities(pid)

@spec capabilities(pid()) :: [String.t()]

Returns the capabilities advertised by the cloud-node on join.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

close(pid)

@spec close(pid()) :: :ok

Closes the WebSocket connection.

connect(pid, protocols)

@spec connect(pid(), [String.t()]) :: :ok | {:error, term()}

Connects to the cloud-node WebSocket and joins the Phoenix channel.

Blocks until the join is acknowledged or returns {:error, reason}.

join_did(pid, did, protocols, did_spec, on_message)

@spec join_did(pid(), String.t(), [String.t()], map() | nil, on_message_fn()) ::
  :ok | {:error, term()}

Joins an additional plugins:<did> topic on the already-open WebSocket.

Blocks until the join is acknowledged. The socket must already be connected (connect/2 returned :ok) and not be mid-reconnect.

  • did — the DID to host; must differ from the primary DID
  • protocols — this topic's payload_types subscription
  • did_spec — merged over the same defaults connect/2 uses (mode, storage, controller, …), or nil
  • on_message — called with the payload of every "message" event on this topic

Returns {:error, :not_connected} before connect/2, {:error, :already_joined} for a DID this channel already hosts, and {:error, :primary_did} for the DID connect/2 joined.

joined_dids(pid)

@spec joined_dids(pid()) :: [String.t()]

Returns the additional DIDs currently joined (excludes the primary DID).

leave_did(pid, did)

@spec leave_did(pid(), String.t()) :: :ok

Leaves an additional DID's topic (phx_leave) and forgets it, so a later reconnect does not bring it back. The socket and every other topic stay up.

No-op for a DID this channel does not host. The primary DID cannot be left this way — use close/1.

send_ack(pid, ids)

@spec send_ack(pid(), [String.t()]) :: :ok

Sends an acknowledgment for a list of message IDs.

Used in legacy mode (cloud nodes without reply protocol).

send_ack_on(pid, target, ids)

@spec send_ack_on(pid(), target(), [String.t()]) :: :ok

send_ack/2 on a specific joined topic — see target/0.

send_fire_and_forget(pid, event, payload)

@spec send_fire_and_forget(pid(), String.t(), map()) :: :ok

Sends a message event fire-and-forget (no reply tracking).

send_fire_and_forget_on(pid, target, event, payload)

@spec send_fire_and_forget_on(pid(), target(), String.t(), map()) :: :ok

send_fire_and_forget/3 on a specific joined topic — see target/0.

send_msg(pid, event, payload)

@spec send_msg(pid(), String.t(), map()) :: {:ok, map()} | {:error, term()}

Sends a message event and waits for a server acknowledgment.

Returns {:ok, reply} where reply is %{status: String.t(), reason: String.t()}, or {:error, reason} on timeout or disconnect.

send_msg_on(pid, target, event, payload)

@spec send_msg_on(pid(), target(), String.t(), map()) ::
  {:ok, map()} | {:error, term()}

send_msg/3 on a specific joined topic — see target/0.

start_link(opts)

@spec start_link(start_opts()) :: GenServer.on_start()

Starts the Channel GenServer.

Options

  • :ws_url — WebSocket URL
  • :api_key — API key for the ?api_key= query parameter
  • :agent_did — agent DID (used as the Phoenix topic plugins:{did})
  • :callbacks — map of callback functions (see module doc)