Amarula.Connection (amarula v0.5.7)

The per-connection process that owns the entire server conversation: the WebSocketClient, the noise cipher (frame encode/decode + counters), IQ correlation, login/handshake, send dispatch, and server-notification handling.

It is also the consumer's endpoint — the pid Amarula.connect/2 returns. Consumer calls (connect, send_text, group_*, …) land here directly, and consumer events go straight to the connection's parent_pid as {:amarula, type, data} (no relay process, no subscriber registry).

Per-send work stays tiny: Connection only frames + writes + correlates (acks, IQ replies). The heavy USync/bundle waits and Signal encrypt run on the per-recipient ConversationSender, which hands back a ready stanza to relay.

Summary

Types

sink()

The consumer event sink — where {:amarula, type, data} events are delivered.

t()

Functions

get_auth_creds(pid)

Current auth creds (carries me.id/me.lid/account once logged in).

send_message(pid, jid, message)

Send a pre-built %Proto.Message{} to jid (1:1 or group). Used for reactions, edits, deletes and media. Returns {:ok, msg_id}.

set_parent(pid, sink)

Re-point the consumer event sink to sink (a sink/0 — pid, registered name, {:via, …}, {name, node}, or nil) on a live connection, without bouncing the websocket. The old sink is demonitored and the new one monitored.

start_link(conn, opts \\ [])

Internal. Starts the bare Connection process only.

update_auth_creds(pid, new_creds)

Updates authentication credentials.

Types

sink()

@type sink() :: pid() | atom() | {:via, module(), term()} | {atom(), node()} | nil

The consumer event sink — where {:amarula, type, data} events are delivered.

A raw pid() is the simplest form, but it is not restart-safe: if the consumer process restarts under a new pid, a pid sink points at a corpse (only set_parent/2 recovers it). Any name GenServer.whereis/1 resolves — a registered atom, a {:via, mod, term} tuple (e.g. Amarula.via(profile) of another registry), or a {name, node} for a remote consumer — re-resolves to the current holder on every event, so it survives the consumer's restart and even this Connection's own restart automatically. nil drops events.

t()

@type t() :: %Amarula.Connection{
  auth_creds: Amarula.Protocol.Auth.AuthUtils.auth_creds() | nil,
  config: Amarula.Protocol.Socket.Types.socket_config(),
  conn: term(),
  connection_state: Amarula.Protocol.Socket.Types.connection_state(),
  connection_timeout_timer: reference() | nil,
  handshake_state:
    Amarula.Protocol.Socket.ConnectionValidator.handshake_state() | nil,
  instance_id: reference() | nil,
  keep_alive_timer: reference() | nil,
  last_error: term() | nil,
  last_recv_time: non_neg_integer() | nil,
  max_retries: non_neg_integer(),
  media_tasks: %{required(reference()) => {GenServer.from(), String.t()}},
  message_counter: non_neg_integer(),
  message_epoch: non_neg_integer(),
  message_tag_prefix: String.t(),
  migrated_pn_sessions: MapSet.t(String.t()),
  msg_retry_counts: %{required(String.t()) => non_neg_integer()},
  noise_state: Amarula.Protocol.Crypto.NoiseHandler.noise_state() | nil,
  parent_monitor: reference() | nil,
  parent_pid: sink(),
  pending_acks: %{
    required(String.t()) =>
      {GenServer.from(), (:ok -> term()), reference(), String.t()}
  },
  pending_iqs: %{required(String.t()) => {atom(), reference()}},
  pending_media_retries: %{
    required(String.t()) => {GenServer.from(), binary(), reference()}
  },
  pending_sends: %{
    required(String.t()) => %{
      msg_id: String.t(),
      text: String.t(),
      target_jid: String.t(),
      devices: [map()]
    }
  },
  qr_refs: [String.t()],
  qr_timer: reference() | nil,
  retry_count: non_neg_integer(),
  retry_delay: non_neg_integer(),
  retry_timer: reference() | nil,
  sender_monitors: %{required(String.t()) => reference()},
  server_response_timeout_timer: reference() | nil,
  task_supervisor: pid() | nil,
  tctoken_recoveries: %{required(reference()) => {:send | :recover, String.t()}},
  waiting_for_server_response: boolean(),
  websocket_client: pid() | nil,
  websocket_monitor: reference() | nil
}

Functions

get_auth_creds(pid)

Current auth creds (carries me.id/me.lid/account once logged in).

send_message(pid, jid, message)

Send a pre-built %Proto.Message{} to jid (1:1 or group). Used for reactions, edits, deletes and media. Returns {:ok, msg_id}.

set_parent(pid, sink)

@spec set_parent(GenServer.server(), sink()) :: :ok

This is the recovery path when the process that called connect/2 restarts while the connection survives: the new consumer re-attaches instead of forcing a stop+reconnect. For a sink that should survive both the consumer's and this connection's restarts, pass a name rather than a raw pid (see sink/0).

start_link(conn, opts \\ [])

Internal. Starts the bare Connection process only.

This starts just this GenServer — not a working connection. A functional connection needs the surrounding per-connection tree (its sender supervisor and the InstanceRegistry wiring its siblings resolve through); without that, the first send fails. ConnectionSupervisor calls this as the tree's child.

Consumers must not put {Amarula.Connection, …} in their own supervisor — that yields a half-wired connection. The supported entry point is Amarula.connect/2, which builds the whole tree (and owns the protocol-driven restarts — e.g. the 515 reconnect after QR pairing — that the consumer should not have to manage). To control naming/distribution, pass a :registry in the connection config (e.g. Horde.Registry); see Amarula.ProfileRegistry.

opts:

:name — registered name (a :via tuple into InstanceRegistry, passed by ConnectionSupervisor). Omit to start an unnamed process.
:parent_pid — process to receive {:amarula, type, data} events

update_auth_creds(pid, new_creds)

Updates authentication credentials.

Called when credentials are updated (e.g., after successful pairing).