Portals.Connection (portals v0.1.0)

Owns one worker's socket, protocol state machine, and in-flight request table (FR-1, FR-2). One Portals.Connection GenServer corresponds to one launched worker process.

Lifecycle: allocates a private Unix-domain socket (or, on platforms without Unix-socket support, a framed stdio port), launches the worker through Portals.WorkerLifecycle.Port, accepts its single connection, and completes the exact-version HELLO/READY handshake — all before replying to start_link/1, so a successfully started connection is always immediately usable.

A worker crash, protocol violation, or handshake failure fails every outstanding request with a terminal %Portals.Error{} exactly once and moves the connection to :closed; it never crashes the caller or the BEAM. Portals.Connection does not restart itself — that is a pool's responsibility (Phase 4).

Summary

Types

start_opts()

Functions

async(conn, module, function, args, opts \\ [])

await(request, timeout \\ 5000)

Block the calling process until request completes or timeout elapses.

call(conn, module, function, args, opts \\ [])

call!(conn, module, function, args, opts \\ [])

cancel(request)

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

health(conn)

open_stream(conn, module, function, args, opts \\ [])

Open a bidirectional stream against module/function (FR-5).

start_link(opts)

stop(conn, deadline_ms \\ 5000)

Types

start_opts()

@type start_opts() :: [
  command: binary(),
  args: [binary()],
  transport: :unix | :stdio,
  limits: Portals.Protocol.limits(),
  connect_timeout: timeout(),
  handshake_timeout: timeout(),
  name: GenServer.name(),
  callback_allowlist: [{module(), atom(), arity()}] | nil,
  message_envelope: :wrapped | :raw
]

Functions

async(conn, module, function, args, opts \\ [])

@spec async(GenServer.server(), binary(), binary(), list(), keyword()) ::
  {:ok, Portals.Request.t()} | {:error, Portals.Error.t()}

await(request, timeout \\ 5000)

@spec await(Portals.Request.t(), timeout()) ::
  {:ok, term()} | {:error, Portals.Error.t()}

Block the calling process until request completes or timeout elapses.

call(conn, module, function, args, opts \\ [])

@spec call(GenServer.server(), binary(), binary(), list(), keyword()) ::
  {:ok, term()} | {:error, Portals.Error.t()}

call!(conn, module, function, args, opts \\ [])

@spec call!(GenServer.server(), binary(), binary(), list(), keyword()) :: term()

cancel(request)

@spec cancel(Portals.Request.t()) :: :ok

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

health(conn)

@spec health(GenServer.server()) :: map()

open_stream(conn, module, function, args, opts \\ [])

@spec open_stream(GenServer.server(), binary(), binary(), list(), keyword()) ::
  {:ok, Portals.Stream.t()} | {:error, Portals.Error.t()}

Open a bidirectional stream against module/function (FR-5).

Dispatches a CALL whose request_id doubles as the stream's stream_id, grants the worker its initial inbound byte credit, and returns a %Portals.Stream{} handle owned by opts[:owner] (defaulting to the calling process). Stream capacity is bounded separately from unary in-flight capacity: it is limited by the max_streams the worker advertised in its HELLO.

start_link(opts)

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

stop(conn, deadline_ms \\ 5000)

@spec stop(GenServer.server(), timeout()) :: :ok