ConduitMcp.Transport.SSEStream (ConduitMCP v0.11.0)

Copy Markdown View Source

Server-Sent Events primitives shared by every stream the HTTP transports hold open: the legacy GET /sse stream of ConduitMcp.Transport.SSE and the modern streamed POST responses (ConduitMcp.StreamSpec).

A stream pins a process, a socket and a Plug.Conn for its whole life, so every stream built on this module is bounded three ways:

  • a deadline — loop/3 returns :deadline once a monotonic deadline passes, checked at the top of every iteration;
  • a connection cap — acquire_slot/3 admits at most max streams per slot table (see "Slot accounting");
  • disconnect detection — every write (send_frame/2, keep_alive/1) reports {:error, reason} once the client is gone.

Frames

  • frame/1 — a JSON-RPC message as an SSE message event: event: message, one data: line, a blank line. JSON escapes every newline inside strings, so the message always fits one data: line.
  • keep_alive_frame/0 — ":\r\n", an SSE comment. Clients discard it; intermediaries see traffic and keep the connection open.

open/1 sends the response head: content-type: text/event-stream, cache-control: no-store and x-accel-buffering: no (so nginx does not buffer the stream), and no connection header — RFC 9113 §8.2.2 forbids connection-specific headers in HTTP/2 and strict clients reject them.

The receive loop

loop/3 waits for messages and keep-alive ticks until a callback halts it, a write fails, or the deadline passes. Keep-alive ticks follow a fixed schedule — every :interval milliseconds since the loop started — so a stream of messages arriving faster than the interval cannot starve them, nor the deadline check.

Every message is received and handed to :on_message, so anything a callback does not care about is drained and the mailbox cannot grow without bound over a long-lived stream. The one exception is {:bandit, _}: under HTTP/2 the plug runs inside Bandit's per-stream process (Bandit.HTTP2.StreamProcess), and the connection process delivers {:send_window_update, delta} and {:rst_stream, code} to this mailbox for chunk/2 to read back with a selective receive. Draining them would discard flow-control credit (eventually a FLOW_CONTROL_ERROR) and ignore stream cancellation — an EventSource.close() sends RST_STREAM, not a TCP close.

Summary

Types

loop_option()

Options for loop/3

outcome()

Why loop/3 returned.

step(state)

What a loop callback returns

Functions

acquire_slot(table, max, sweep_window_ms \\ 1000)

Takes a slot for the calling process in slot table table, admitting at most max live streams. Returns false at the cap — and when the table is missing or unusable, because a resource cap must read "cannot count" as "no".

active_slots(table)

The number of slot rows in table whose process is alive. Read-only: a getter that swept dead rows would hide leaks from the tests that should see them.

ensure_slot_table(table)

Creates named slot table table unless it exists. A fallback for embedding contexts where the :conduit_mcp application is not started; normally the table's owner created it at boot.

frame(message)

Encodes a JSON-RPC message as one SSE message event.

keep_alive(conn)

Writes keep_alive_frame/0. {:error, reason} means the client is gone.

keep_alive_frame()

The keep-alive frame, ":\r\n": an SSE comment line.

loop(conn, state, opts)

Runs the receive loop described in the moduledoc and returns why it stopped, with the final conn and state.

open(conn)

Starts the SSE response: the stream headers, then send_chunked(conn, 200).

release_slot(table)

Releases the calling process's slot in table, if it holds one. Idempotent; :ok when the table is gone (the slot went with it).

send_frame(conn, frame)

Writes an already-encoded frame. {:error, reason} means the client is gone.

slot_table_opts()

The options every slot table is created with.

streams_table()

The slot table behind streamed POST responses (:max_streams), owned by ConduitMcp.Transport.SSEStream.Owner.

Types

loop_option()

@type loop_option() ::
  {:interval, pos_integer()}
  | {:deadline, integer()}
  | {:on_message, (term(), Plug.Conn.t(), term() -> step(term()))}
  | {:on_tick, (Plug.Conn.t(), term() -> step(term()))}

Options for loop/3:

  • :interval (required) — milliseconds between keep-alive ticks.
  • :deadline (required) — the System.monotonic_time(:millisecond) at which the loop returns :deadline.
  • :on_message — (message, conn, state -> step), called for every message except {:bandit, _}. Default: drain and continue.
  • :on_tick — (conn, state -> step), called on every keep-alive tick. Default: write keep_alive_frame/0, :closed when that fails.

outcome()

@type outcome() :: :halt | :closed | :deadline

Why loop/3 returned.

step(state)

@type step(state) ::
  {:cont, Plug.Conn.t(), state}
  | {:halt, Plug.Conn.t(), state}
  | {:closed, Plug.Conn.t(), state}

What a loop callback returns:

  • {:cont, conn, state} — keep looping;
  • {:halt, conn, state} — stop; loop/3 returns :halt;
  • {:closed, conn, state} — a write failed: the client is gone; loop/3 returns :closed.

Functions

acquire_slot(table, max, sweep_window_ms \\ 1000)

@spec acquire_slot(atom() | :ets.tid(), pos_integer(), non_neg_integer()) :: boolean()

Takes a slot for the calling process in slot table table, admitting at most max live streams. Returns false at the cap — and when the table is missing or unusable, because a resource cap must read "cannot count" as "no".

sweep_window_ms is how long a sweep that found no dead rows suppresses the next one (default 1000).

active_slots(table)

@spec active_slots(atom() | :ets.tid()) :: non_neg_integer()

The number of slot rows in table whose process is alive. Read-only: a getter that swept dead rows would hide leaks from the tests that should see them.

ensure_slot_table(table)

@spec ensure_slot_table(atom()) :: :ok

Creates named slot table table unless it exists. A fallback for embedding contexts where the :conduit_mcp application is not started; normally the table's owner created it at boot.

frame(message)

@spec frame(map()) :: iodata()

Encodes a JSON-RPC message as one SSE message event.

Raises if message is not JSON-encodable, so callers building frames in user code (ConduitMcp.Progress, ConduitMcp.Log) fail where the bad value was produced.

Example

iex> ConduitMcp.Transport.SSEStream.frame(%{"jsonrpc" => "2.0", "method" => "ping"})
...> |> IO.iodata_to_binary()
~s(event: message\ndata: {"jsonrpc":"2.0","method":"ping"}\n\n)

keep_alive(conn)

@spec keep_alive(Plug.Conn.t()) :: {:ok, Plug.Conn.t()} | {:error, term()}

Writes keep_alive_frame/0. {:error, reason} means the client is gone.

keep_alive_frame()

@spec keep_alive_frame() :: String.t()

The keep-alive frame, ":\r\n": an SSE comment line.

loop(conn, state, opts)

@spec loop(Plug.Conn.t(), state, [loop_option()]) :: {outcome(), Plug.Conn.t(), state}
when state: term()

Runs the receive loop described in the moduledoc and returns why it stopped, with the final conn and state.

open(conn)

@spec open(Plug.Conn.t()) :: Plug.Conn.t()

Starts the SSE response: the stream headers, then send_chunked(conn, 200).

release_slot(table)

@spec release_slot(atom() | :ets.tid()) :: :ok

Releases the calling process's slot in table, if it holds one. Idempotent; :ok when the table is gone (the slot went with it).

send_frame(conn, frame)

@spec send_frame(Plug.Conn.t(), iodata()) :: {:ok, Plug.Conn.t()} | {:error, term()}

Writes an already-encoded frame. {:error, reason} means the client is gone.

slot_table_opts()

@spec slot_table_opts() :: [term()]

The options every slot table is created with.

streams_table()

@spec streams_table() :: atom()

The slot table behind streamed POST responses (:max_streams), owned by ConduitMcp.Transport.SSEStream.Owner.