Opens a long-lived SSE stream and owns its receive loop.
Dstar.Page streams run on this module; plain controllers use it the same
way. It owns the parts of a stream's lifecycle that are easy to get subtly
wrong in a hand-rolled receive:
- Keyed claims. With
key:,open/2claims{key, tabId}through the opt-inDstar.Utility.StreamRegistrybefore SSE starts, and reports a refused claim as{:error, conn}. - Takeover. When a newer stream for the same tab claims the key, the
registry signals this stream with its exact claim generation.
run/2recognises only its own generation, releases the claim, then tears down. Stale signals from an earlier stream on a reused keep-alive process are ignored. - Adapter plumbing.
{:plug_conn, :sent}is skipped, and Bandit's{:bandit, _}flow-control messages are left in the mailbox for Bandit's own selective receive. A catch-allreceivewould swallow them and can stall an HTTP/2 stream. - Liveness and release. An idle stream is probed every
:idle_checkms and torn down once the client is gone. The claim is always released before:disconnectruns, and again if any callback raises.
Example
def stream(conn, _params) do
user = conn.assigns.current_user
case Dstar.Stream.open(conn, key: user.id) do
{:ok, conn} ->
Dstar.Stream.run(conn,
connect: fn conn ->
Phoenix.PubSub.subscribe(MyApp.PubSub, "user:#{user.id}")
conn
end,
info: fn
{:notification, n}, conn -> Dstar.Elements.append(conn, row(n), "#inbox")
:logout, conn -> {:halt, conn}
end,
disconnect: fn _conn ->
Phoenix.PubSub.unsubscribe(MyApp.PubSub, "user:#{user.id}")
end
)
{:error, conn} ->
# 400/413 (bad signals) or 503 (claim refused): plain HTTP, no SSE.
conn
end
endDstar.start_stream/2,3 and Dstar.Utility.StreamRegistry.release/1
remain available for loops that must own their receive.
Summary
Types
@type info_result() :: Plug.Conn.t() | {:halt, Plug.Conn.t()}
Functions
@spec open(Plug.Conn.t(), keyword()) :: {:ok, Plug.Conn.t()} | {:error, Plug.Conn.t()}
Starts SSE on conn, claiming a per-tab key first when :key is given.
Returns {:ok, conn} with a chunked SSE conn, or {:error, conn} with an
ordinary halted HTTP response already sent: 400/413 when a keyed stream's
signals cannot be read, 503 when the registry refuses the claim. Never
start subscriptions on the error path.
Without :key, this is Dstar.start/1. With :key, it is
Dstar.start_stream/3: a missing or invalid tabId signal falls back to an
ordinary unkeyed stream.
Options
:key— scope key for per-tab deduplication (any term, e.g.user.id). RequiresDstar.Utility.StreamRegistryin the supervision tree.:max_bytes— limit for reading signals on a keyed stream (seeDstar.Signals.fetch/2).
@spec run(Plug.Conn.t(), keyword()) :: Plug.Conn.t()
Runs the receive loop for a conn returned by open/2, until the client
disconnects, a callback halts, or a newer stream takes over the key.
Returns the final conn. The stream's claim is released before
:disconnect runs, and on the way out if any callback raises.
Options
:connect—(conn -> conn), called once before the loop. Subscribe here: a raise releases the claim.:info—(message, conn -> conn | {:halt, conn}), called for every application message.{:halt, conn}ends the stream. Without it, messages are logged and dropped.:replaced—(message, conn -> conn), called when a newer stream takes over this key, after the claim is released and before:disconnect.messageis{:EXIT, pid, :replaced}.:disconnect—(conn -> any), called once when the loop ends for any reason other than a raise.:idle_check— ms without messages before probing the connection (default 30000).