Class: Insika::Server::SSEBody

Inherits:
Object
  • Object
show all
Defined in:
lib/insika/server/sse_body.rb

Overview

SSE response body (evolves SSEStream). SSEStream received a producer block (the Runner wrote into it); SSEBody DRAINS a Subscription from the EventStream: each subscriber has its own queue; #each blocks the CONSUMER's fiber until the subscription closes. The wire is EXACTLY Event#to_h.

Constant Summary collapse

PING =

SSE comment — doesn't pollute the consumer

": ping\n\n"
DEFAULT_SERIALIZE =

serialize: maps an Event -> String (SSE frame) OR nil (discarded event, no frame). Default = the canonical wire data: <Event#to_h>. The /v1/responses adapter injects a serializer that produces OpenAI Responses events (and skips those with no counterpart).

->(event) { "data: #{JSON.generate(event.to_h)}\n\n" }

Instance Method Summary collapse

Constructor Details

#initialize(subscription:, heartbeat: 15, serialize: nil) ⇒ SSEBody

subscription: any object with #each (yields Events) and #close. heartbeat: seconds of silence before emitting a ping (15s clears 60s ALB/nginx idle timeouts with room to spare).



27
28
29
30
31
 
# File 'lib/insika/server/sse_body.rb', line 27

def initialize(subscription:, heartbeat: 15, serialize: nil)
  @subscription = subscription
  @heartbeat = heartbeat
  @serialize = serialize || DEFAULT_SERIALIZE
end

Instance Method Details

#call(stream) ⇒ Object

Rack 3 STREAMING BODY (#call(stream)), NOT #each. Under protocol-rack/protocol-http1 (the stack of Async::HTTP::Server AND Falcon), a body that responds to #each is routed to Body::Enumerable, whose read runs the #each in a PLAIN Enumerator Fiber (not an Async::Task) — there Async::Task.current raises "No async task available", the loop died swallowed in the rescue and the body came out EMPTY. By exposing #call (and NOT #each), the body is routed to Body::Streaming, which schedules the block via Fiber.schedule under the reactor's scheduler — so the subscription drains and the frames actually reach the socket (incrementally).

The stream (Protocol::HTTP::Body::Stream) responds to #write/#close. Drains the subscription DIRECTLY (no Async::Task.current): when this fiber blocks waiting for the next event, the scheduler runs the writer, which pushes the already-written frame to the socket. Heartbeat via Timeout.timeout (scheduler hook), which works in the scheduled fiber — keeps the connection alive while idle (L4).



48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/insika/server/sse_body.rb', line 48

def call(stream)
  drain(stream)
rescue StandardError
  # Client disconnected: `stream.write` raises when the socket closes.
  # No exception escapes; the turn's task is NEVER cancelled here — the
  # execution belongs to the runtime, not the connection (reconnect at /v1/events).
  nil
ensure
  @subscription.close
  stream.close
end