# `ConduitMcp.Transport.SSEStream`
[🔗](https://github.com/nyo16/conduit_mcp/blob/v0.11.0/lib/conduit_mcp/transport/sse_stream.ex#L1)

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.

# `loop_option`

```elixir
@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`

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

Why `loop/3` returned.

# `step`

```elixir
@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`.

# `acquire_slot`

```elixir
@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`

```elixir
@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`

```elixir
@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`

```elixir
@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`

```elixir
@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`

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

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

# `loop`

```elixir
@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`

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

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

# `release_slot`

```elixir
@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`

```elixir
@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`

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

The options every slot table is created with.

# `streams_table`

```elixir
@spec streams_table() :: atom()
```

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

---

*Consult [api-reference.md](api-reference.md) for complete listing*