ALLM.Providers.Support.WebSocket.InputLoop (allm v0.6.1)

Copy Markdown View Source

The owner-side loop shared by the WebSocket input-stream adapters: a socket that speaks while an ALLM.Providers.Support.InputPump feeds it.

Layer B helper. An adapter runs one Stream.resource/3 in the reducing process, which owns the socket, and builds it on these functions over a state map made by loop_state/3 (the adapter adds its own keys):

  • start_pump/3 starts the input pump, after the connect and whatever the adapter must see first, so a refused session never reduces the input;
  • next_message/1 selects the next pump or transport message of this stream, and nothing else in the mailbox, with a silence deadline and optional keep-alive and wake_at deadlines;
  • timed_out?/1 tells, after :wake, whether the silence deadline passed;
  • handle_transport/2 and send_json/2 wrap the transport;
  • stop_pump/1 and close_loop/1 (the after function) release everything.

The adapter keeps its payload handlers and builds its own errors: every failure here returns the transport's raw cause.

Timers

stream_timeout is the silence allowed before timed_out?/1 holds, and both a transport message and a pump message reset it, so a slow input does not time out a socket whose server is waiting, and a quiet server does not time out while input still flows. keepalive_ms is how long no client frame may go out, while input still flows, before next_message/1 wakes the adapter to send one; :infinity turns the keep-alive off, for a protocol that has none. wake_at, when the adapter sets it to a System.monotonic_time(:millisecond) instant, is a third deadline of its own (for example a bounded hold on an event): no message resets it, next_message/1 returns :wake once it passes, and the adapter clears it back to nil when done with it. After such a wake, timed_out?/1 still tells whether the silence deadline passed as well.

Summary

Types

t()

The loop state: these keys plus any the adapter adds.

Functions

close_loop(state)

The Stream.resource/3 after function: close the socket, remove its messages from the mailbox, and stop the pump.

handle_transport(state, message)

Interpret a transport message selected by next_message/1, keeping the connection the transport returns.

loop_state(ws, stream_timeout, keepalive_ms)

A fresh loop state over the transport module ws, with no connection and no pump yet and no wake_at deadline. stream_timeout and keepalive_ms are in milliseconds, or :infinity.

next_message(state)

The next message of this stream: a pump message (classified by ALLM.Providers.Support.InputPump.classify/2) or a transport message, with the silence clock reset, or :wake when the silence, keep-alive or wake_at deadline passed first. Nothing else in the mailbox is consumed.

send_json(state, message)

Send message JSON-encoded as one text frame, restarting the keep-alive clock. The error's third element is the transport's raw cause.

start_pump(state, input, opts)

Start the input pump over input, owned by the calling process, with opts[:adapter_opts][:input_window] (default ALLM.Providers.Support.InputPump.default_window/0) as its credit window.

stop_pump(state)

Stop the pump, if one runs, and drain its messages. Idempotent.

timed_out?(state)

After :wake: whether the silence deadline passed (the stream ends with a timeout) rather than the keep-alive one (the adapter sends its keep-alive) or the adapter's own wake_at deadline (the adapter acts on it, e.g. releases a held segment). More than one can have passed; an adapter that must not report a timeout for work the wake_at deadline completes checks that first.

Types

t()

@type t() :: %{
  :ws => module(),
  :conn => ALLM.Providers.Support.WebSocket.conn() | nil,
  :pump => {pid(), reference()} | nil,
  :stream_timeout => timeout(),
  :keepalive_ms => timeout(),
  :last_activity => integer(),
  :last_sent => integer(),
  :input_done? => boolean(),
  :wake_at => integer() | nil,
  optional(atom()) => term()
}

The loop state: these keys plus any the adapter adds.

Functions

close_loop(state)

@spec close_loop(t()) :: :ok

The Stream.resource/3 after function: close the socket, remove its messages from the mailbox, and stop the pump.

handle_transport(state, message)

@spec handle_transport(t(), term()) ::
  {:ok, t(), [ALLM.Providers.Support.WebSocket.frame() | :closed]}
  | :unknown
  | {:error, t(), term()}

Interpret a transport message selected by next_message/1, keeping the connection the transport returns.

loop_state(ws, stream_timeout, keepalive_ms)

@spec loop_state(module(), timeout(), timeout()) :: t()

A fresh loop state over the transport module ws, with no connection and no pump yet and no wake_at deadline. stream_timeout and keepalive_ms are in milliseconds, or :infinity.

next_message(state)

@spec next_message(t()) ::
  {:pump, ALLM.Providers.Support.InputPump.classified(), t()}
  | {:transport, term(), t()}
  | :wake

The next message of this stream: a pump message (classified by ALLM.Providers.Support.InputPump.classify/2) or a transport message, with the silence clock reset, or :wake when the silence, keep-alive or wake_at deadline passed first. Nothing else in the mailbox is consumed.

send_json(state, message)

@spec send_json(t(), term()) :: {:ok, t()} | {:error, t(), term()}

Send message JSON-encoded as one text frame, restarting the keep-alive clock. The error's third element is the transport's raw cause.

start_pump(state, input, opts)

@spec start_pump(t(), Enumerable.t(), keyword()) :: t()

Start the input pump over input, owned by the calling process, with opts[:adapter_opts][:input_window] (default ALLM.Providers.Support.InputPump.default_window/0) as its credit window.

stop_pump(state)

@spec stop_pump(t()) :: t()

Stop the pump, if one runs, and drain its messages. Idempotent.

timed_out?(state)

@spec timed_out?(t()) :: boolean()

After :wake: whether the silence deadline passed (the stream ends with a timeout) rather than the keep-alive one (the adapter sends its keep-alive) or the adapter's own wake_at deadline (the adapter acts on it, e.g. releases a held segment). More than one can have passed; an adapter that must not report a timeout for work the wake_at deadline completes checks that first.