Streaming & SSE

A controller returns sse(req) or stream(req, content_type) with a block. The server holds the connection open and sends each chunk over Transfer-Encoding: chunked as the block calls out.emit(...) / out.write(...) — ideal for AI token streaming, progress, live dashboards, and large CSV/JSON exports.

sse(req) do |out| ... end sends text/event-stream. Call out.emit(data, event?) per event; the browser's EventSource receives them live.

def stream(req)
  sse(req) do |out|
    for item in Notification.live()
      out.emit(item.to_json, "notice")   # event: notice\n data: {...}
    end
    out.emit("done")                      # a plain data: event
  end
end

const es = new EventSource("/feed/stream")
es.addEventListener("notice", e => console.log(JSON.parse(e.data)))

out.emit returns false when the client has disconnected — check it to stop an open-ended loop early.

stream(req, content_type) do |out| ... end sends a raw chunked body. Use out.write(chunk) — no SSE framing — to stream a file or report without buffering it in memory.

def export(req)
  stream(req, "text/csv") do |out|
    out.write("name,score\n")
    for row in Player.order("score desc").each_row()
      out.write(row.name + "," + str(row.score) + "\n")
    end
  end
end

Named emit, not send, because send is the universal metaprogramming method.

# Stream an answer over SSE — tokens reach the browser as they generate.
def ask(req)
  sse(req) do |out|
    answer = out.llm_stream("You are concise.", req["query"]["q"])
    ChatLog.create({ "q": req["query"]["q"], "a": answer })
  end
end

Needs an LLM configured (SOLI_LLM_API_KEY / SOLI_LLM_URL). To stream an answer grounded in your data, retrieve context first with Model.rag / Model.similar, then build the prompt and call out.llm_stream.

The sse(...)/stream(...) blocks above hold a worker thread while they run — great for a finite job, but you don't want one worker per idle connection on a dashboard with thousands of viewers. For that, subscribe instead: sse_subscribe(req, topic) registers the connection and returns immediately (no worker held), and sse_broadcast(topic, data, event?) fans an event out to every subscriber — called from any controller, job, or model callback.

# Each browser holds a cheap async connection — not a worker.
def subscribe(req)
  sse_subscribe(req, "user:#{current_user.id}")
end

# Push from anywhere — a controller action, a background job, a callback.
def notify(req)
  reached = sse_broadcast("user:#{params["id"]}", params["msg"], "alert")
  render_json({ "delivered": reached })
end

const es = new EventSource("/notifications/subscribe")
es.addEventListener("alert", e => toast(e.data))

A subscription costs an async task, not a thread — one worker can hold thousands of live subscribers. Disconnected clients are pruned on the next broadcast; sse_subscribers(topic) returns the current count.

broadcast(channel, payload) fans payload out to both the WebSocket channel and the SSE topic of the same name — so a page listens over whichever transport it uses and you publish once. Non-string payloads auto-serialize to JSON; it returns the SSE subscriber count. Call it from any controller, job, or model callback.

def create(req)
  post = Post.create(permit(params, {"title": true}))
  # Reaches WS clients in room "posts" AND SSE subscribers of topic "posts".
  broadcast("posts", { "event": "created", "id": post.id, "title": post.title })
  redirect("/posts/#{post.id}")
end

Models carry a shortcut: Post.broadcast(payload) publishes to the model's collection channel ("posts") — ideal in an after_save callback so every write pushes a change event to subscribed clients. This is a general pub/sub primitive; for a LiveView that re-renders on writes, prefer reactive live queries instead.