Class: Xeno::StreamsController

Inherits:
ApiController
  • Object
show all
Includes:
ActionController::Live
Defined in:
app/controllers/xeno/streams_controller.rb

Overview

The session event stream: SSE by default, NDJSON on request (?format=ndjson or Accept: application/x-ndjson — one envelope per line, friendlier to curl/jq and non-browser consumers). ?start_index=N rewinds/resumes — events are durable rows, so reconnecting clients replay history and then follow live. The stream ends when the session reaches a terminal status (or the optional configured max duration).

Known limitation (documented): ActionController::Live holds a thread per connected client.

Defined Under Namespace

Classes: NdjsonWriter, SseWriter

Constant Summary collapse

KEEPALIVE_INTERVAL =

seconds without a write before a ping

5

Constants inherited from ApiController

ApiController::DEV_PRINCIPAL

Instance Method Summary collapse

Instance Method Details

#show ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'app/controllers/xeno/streams_controller.rb', line 42

def show
  session = find_owned_session!
  writer = build_writer
  response.headers["Content-Type"] = writer.content_type
  response.headers["Last-Modified"] = Time.now.httpdate # disable buffering middlewares

  cursor = params.fetch(:start_index, 0).to_i
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  last_write = started

  batch_limit = Xeno.config.stream_catch_up_batch
  loop do
    events = session.events.where(index: cursor..).order(:index).limit(batch_limit).to_a
    events.each do |event|
      writer.event(event)
      cursor = event.index + 1
    end
    last_write = Process.clock_gettime(Process::CLOCK_MONOTONIC) if events.any?

    # A full batch means more history is waiting — keep paging through
    # the catch-up without sleeping or ending on a terminal status.
    next if events.size == batch_limit

    break unless session.reload.active?
    break if stream_expired?(started)
    # A graceful stop must not wait out the in-flight-request window —
    # the stream is resumable by design (events are durable rows, the
    # client reconnects with its cursor), so close it and let the
    # server exit. The 5s force cap in the puma config stays as the
    # backstop.
    break if server_shutting_down?

    # A quiet stream never writes, so a dead client would never raise and
    # this thread would poll forever — keepalives make disconnects visible.
    if Process.clock_gettime(Process::CLOCK_MONOTONIC) - last_write > KEEPALIVE_INTERVAL
      writer.keepalive
      last_write = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    end

    sleep Xeno.config.stream_poll_interval
  end
rescue ActionController::Live::ClientDisconnected, IOError
  # the client went away — nothing to clean up, events are durable
ensure
  writer&.close
end