Functions for reading and patching Datastar signals via SSE.
{:ok, signals, conn} = Dstar.Signals.fetch(conn)
conn |> patch(%{count: 42, message: "Hello"})
conn |> patch(%{count: 42}, only_if_missing: true)
conn |> remove_signals("user.session")Summary
Types
Why a signal payload could not be fetched
Functions
Safely fetches signals and returns the connection that owns the body state.
Formats a signals patch as an SSE event string (for stateless responses).
Formats a signal removal as an SSE event string (for stateless responses).
Signals that a collection changed, without saying how.
Patches signals on the client by sending an SSE event.
Patches signals using a raw JSON string.
Reads signals only when the relevant params have already been fetched.
Removes signals from the client by setting them to nil.
Sends a controlled plain-text response for a fetch/2 error.
Types
@type fetch_error() :: :malformed | :not_an_object | :too_large | {:read_body, term()}
Why a signal payload could not be fetched
Functions
@spec fetch(Plug.Conn.t(), keyword()) :: {:ok, map(), Plug.Conn.t()} | {:error, fetch_error(), Plug.Conn.t()}
Safely fetches signals and returns the connection that owns the body state.
Only JSON objects are accepted. GET and DELETE read the datastar query
parameter; other methods use already-fetched body params or read the raw
JSON body. Missing signals or an empty raw body produce an empty object;
an explicitly empty datastar= query value is malformed.
Raw reads bound both Plug.Conn.read_body/2's :length and
:read_length. The default maximum is 1000000 bytes and can
be changed per call with :max_bytes. When a parser already populated
body_params, that parser owns raw-body limits; configure it to the same
or a smaller limit.
The returned conn must be threaded forward:
case Dstar.Signals.fetch(conn, max_bytes: 64_000) do
{:ok, signals, conn} -> handle(conn, signals)
{:error, reason, conn} -> Dstar.Signals.send_error(conn, reason)
endOptions
:max_bytes— maximum raw JSON payload size (default 1000000)
Formats a signals patch as an SSE event string (for stateless responses).
Example
format_patch(%{count: 42})
# => "event: datastar-patch-signals\ndata: signals {\"count\":42}\n\n"Formats a signal removal as an SSE event string (for stateless responses).
Examples
format_remove("user.profile")
# => "event: datastar-patch-signals\ndata: signals {\"user\":{\"profile\":null}}\n\n"
format_remove(["user.a", "user.b"])
# => "event: datastar-patch-signals\ndata: signals {\"user\":{\"a\":null,\"b\":null}}\n\n"@spec nudge(Plug.Conn.t(), String.t() | atom(), keyword()) :: Plug.Conn.t()
Signals that a collection changed, without saying how.
Patches nudges.<key> with a fresh integer. Tabs watching that signal
(see Dstar.Actions.on_nudge/2) re-run their own load action, which
carries their current filter/sort/page signals — so every tab gets a
view correct for that tab. This is the default for any list that is
filtered, sorted or paginated; the stream is one-way and cannot learn a
tab's view state after connect. See the
Live collections guide.
The value is System.unique_integer([:positive, :monotonic]). The client
only fires its handler when a signal's value actually changes, so
re-sending a constant would be a silent no-op.
key is a single signal path segment: ~r/^[a-zA-Z0-9_]+$/.
Example
def handle_info({:posts_changed, _}, conn), do: nudge(conn, "posts")@spec patch(Plug.Conn.t(), map(), keyword()) :: Plug.Conn.t()
Patches signals on the client by sending an SSE event.
Options
:only_if_missing- Only patch signals that don't exist on the client (default: false):event_id- Event ID for client tracking:retry- Retry duration in milliseconds
Example
conn
|> Dstar.Signals.patch(%{count: 42})
|> Dstar.Signals.patch(%{message: "Hello"}, only_if_missing: true)@spec patch_raw(Plug.Conn.t(), String.t(), keyword()) :: Plug.Conn.t()
Patches signals using a raw JSON string.
Multi-line JSON is sent as one data: signals <line> per line, which the
client reassembles into a single payload.
Example
conn
|> Dstar.Signals.patch_raw(~s({"count": 42}))@spec read(Plug.Conn.t()) :: map()
Reads signals only when the relevant params have already been fetched.
This convenience API cannot safely own a raw request body because it cannot
return Plug's updated conn. Use fetch/2 when body_params (or GET/DELETE
query_params) are unfetched. Invalid JSON and non-object JSON raise; code
that needs a controlled 400/413 response should use fetch/2.
@spec remove_signals(Plug.Conn.t(), String.t() | [String.t()], keyword()) :: Plug.Conn.t()
Removes signals from the client by setting them to nil.
Accepts a single dot-notated path string or a list of paths.
Paths are converted to a nested map with nil values,
then passed to patch/3.
Examples
# Remove a single signal
conn |> remove_signals("user.profile.theme")
# Remove multiple signals with shared prefix
conn |> remove_signals(["user.name", "user.email"])
# Remove top-level signal
conn |> remove_signals("count")Options
:only_if_missing- Only remove if signal doesn't exist (default: false):event_id- Event ID for client tracking:retry- Retry duration in milliseconds
Path validation
Raises ArgumentError on a path that is empty, starts or ends with a dot,
or contains consecutive dots — each would otherwise produce a signal name
with an empty segment.
@spec send_error(Plug.Conn.t(), fetch_error()) :: Plug.Conn.t()
Sends a controlled plain-text response for a fetch/2 error.
Oversized payloads receive 413; every other input failure receives 400. This is safe to call before starting SSE.