Dstar.Stream (dstar v0.3.0)

Copy Markdown View Source

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/2 claims {key, tabId} through the opt-in Dstar.Utility.StreamRegistry before 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/2 recognises 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-all receive would swallow them and can stall an HTTP/2 stream.
  • Liveness and release. An idle stream is probed every :idle_check ms and torn down once the client is gone. The claim is always released before :disconnect runs, 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
end

Dstar.start_stream/2,3 and Dstar.Utility.StreamRegistry.release/1 remain available for loops that must own their receive.

Summary

Functions

Starts SSE on conn, claiming a per-tab key first when :key is given.

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.

Types

info_result()

@type info_result() :: Plug.Conn.t() | {:halt, Plug.Conn.t()}

Functions

open(conn, opts \\ [])

@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

run(conn, opts \\ [])

@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. message is {: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).