Class: Smplkit::EventStream

Inherits:
Object
  • Object
show all
Defined in:
lib/smplkit/event_stream.rb

Overview

Manages the single live-updates event stream to the app service.

A single EventStream instance is shared across all product modules (config, flags, logging) within one Smplkit::Client. Product modules register listeners for specific event names; the shared stream dispatches incoming events to the appropriate listeners. Modules also register refetch callbacks, invoked after every successful reconnect so their caches recover anything the server published while the stream was down.

The stream runs on a dedicated SDK-owned thread that hosts the Async reactor and the underlying async-http I/O. Public methods are thread-safe and non-blocking.

Wire protocol — Server-Sent Events (SSE) over plain HTTPS:

- +GET <app_base_url>/api/v1/events+ with +Accept: text/event-stream+
and +Authorization: Bearer <api_key>+.
- A 200 response with a +text/event-stream+ content type is a
successful connect; auth failure is a plain HTTP 401.
- Each SSE frame carries the event name in the +event:+ field and a
JSON object in +data:+ (+{"id": "<key>"}+ for single-resource
events, +{}+ for bulk refreshes and the initial +connected+ event).
- The server emits a +: keepalive+ comment frame every 30 seconds when
idle; any received bytes count as liveness. Reads that stall past
+READ_TIMEOUT+ seconds tear the connection down for a reconnect.
- The server's +retry:+ field seeds the reconnect backoff base.

On disconnect the reactor reconnects with exponential backoff (base delay doubling up to MAX_BACKOFF seconds), resetting to the base on every successful connect. stop flips @closed and waits; the reader re-checks the flag at least every POLL_INTERVAL, tears its own connection down in-reactor, and the daemon thread terminates.

Defined Under Namespace

Classes: Parser

Constant Summary collapse

READ_TIMEOUT =

Seconds without any bytes from the server (events or keepalive comments) before the connection is considered dead — two missed 30-second server keepalives.

45
POLL_INTERVAL =

How long a single blocking read may park the reactor before it wakes to re-check @closed. Liveness is tracked as a deadline across polls (see read_loop), so this changes nothing on the wire — it exists so stop never has to interrupt the reactor from a foreign thread: a cross-thread close cannot wake a fiber blocked in body.read, which left teardown hanging until the next keepalive.

1.0
MAX_BACKOFF =

Ceiling for the exponential reconnect backoff, in seconds.

60
DEFAULT_RETRY =

Initial reconnect backoff base, in seconds, used until the server supplies its own via the SSE retry: field.

1.0
USER_AGENT =

Sent on the stream request — the platform WAF rejects requests that carry no User-Agent. There is no caller-supplied header surface on the event stream, so the SDK default always applies.

Smplkit.user_agent.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(app_base_url:, api_key:, metrics: nil) ⇒ EventStream

Returns a new instance of EventStream.



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/smplkit/event_stream.rb', line 192

def initialize(app_base_url:, api_key:, metrics: nil)
  @app_base_url = app_base_url
  @api_key = api_key
  @metrics = metrics
  @listeners = Hash.new { |h, k| h[k] = [] }
  @refetch_callbacks = []
  @listeners_lock = Mutex.new
  @connection_status = "disconnected"
  @closed = false
  @stream_thread = nil
  @client = nil
  @response = nil
  @connection_lock = Mutex.new
  @retry_base = DEFAULT_RETRY
  @attempt = 0
  @ever_connected = false
end

Instance Attribute Details

#connection_status ⇒ Object (readonly)

----- Connection status ----------------------------------------



249
250
251
 
# File 'lib/smplkit/event_stream.rb', line 249

def connection_status
  @connection_status
end

Instance Method Details

#build_events_url ⇒ Object

----- URL builder ----------------------------------------------



287
288
289
290
291
 
# File 'lib/smplkit/event_stream.rb', line 287

def build_events_url
  url = @app_base_url.dup
  url = "https://#{url}" unless url.start_with?("https://", "http://")
  "#{url.chomp("/")}/api/v1/events"
end

#dispatch(event_name, payload) ⇒ Object

Dispatch payload to every listener registered for event_name. Event names nothing subscribed to dispatch to zero listeners — unknown events are ignored by construction. Listener exceptions are caught and logged; one bad listener never blocks the rest.



238
239
240
241
242
243
244
245
 
# File 'lib/smplkit/event_stream.rb', line 238

def dispatch(event_name, payload)
  callbacks = @listeners_lock.synchronize { @listeners[event_name].dup }
  callbacks.each do |cb|
    cb.call(payload)
  rescue StandardError => e
    Smplkit.debug("events", "listener for #{event_name} raised: #{e.class}: #{e.message}")
  end
end

#handle_event(event_name, data) ⇒ Object

Process one parsed SSE event the way the live read loop does: parse the JSON payload and dispatch it to the listeners registered for the event name.

Returns :dispatched or :unparseable for the caller to log/observe; the live read loop ignores the return value.



301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
 
# File 'lib/smplkit/event_stream.rb', line 301

def handle_event(event_name, data)
  payload =
    begin
      JSON.parse(data)
    rescue JSON::ParserError
      nil
    end
  unless payload.is_a?(Hash)
    Smplkit.debug("events", "ignoring #{event_name.inspect} event with non-object payload")
    return :unparseable
  end

  dispatch(event_name, payload)
  :dispatched
end

#off(event_name, callback) ⇒ Object 



216
217
218
 
# File 'lib/smplkit/event_stream.rb', line 216

def off(event_name, callback)
  @listeners_lock.synchronize { @listeners[event_name].delete(callback) }
end

#off_reconnect(callback) ⇒ Object 



230
231
232
 
# File 'lib/smplkit/event_stream.rb', line 230

def off_reconnect(callback)
  @listeners_lock.synchronize { @refetch_callbacks.delete(callback) }
end

#on(event_name, &callback) ⇒ Object

----- Listener registration ------------------------------------



212
213
214
 
# File 'lib/smplkit/event_stream.rb', line 212

def on(event_name, &callback)
  @listeners_lock.synchronize { @listeners[event_name] << callback }
end

#on_reconnect(callback = nil, &block) ⇒ Object

Register a refetch callback, invoked (with no arguments) after every successful reconnect — never on the initial connect. Product modules use this to run their bulk-refresh path so caches recover events missed while the stream was down.



224
225
226
227
228
 
# File 'lib/smplkit/event_stream.rb', line 224

def on_reconnect(callback = nil, &block)
  cb = callback || block
  @listeners_lock.synchronize { @refetch_callbacks << cb }
  cb
end

#start ⇒ Object

----- Lifecycle ------------------------------------------------



253
254
255
256
257
258
259
260
261
 
# File 'lib/smplkit/event_stream.rb', line 253

def start
  return if @stream_thread&.alive?

  Smplkit.debug("events", "starting shared event stream background thread")
  @closed = false
  @connection_status = "connecting"
  @stream_thread = Thread.new { run_reactor }
  @stream_thread.name = "smplkit-events" if @stream_thread.respond_to?(:name=)
end

#stop ⇒ Object 



263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/smplkit/event_stream.rb', line 263

def stop
  Smplkit.debug("events", "stopping shared event stream")
  @closed = true
  thread = @stream_thread
  @stream_thread = nil
  if thread
    # The reactor re-checks +@closed+ at least every POLL_INTERVAL and
    # closes its own connection in-reactor on the way out — closing it
    # from this thread instead would race the reactor and cannot wake
    # a blocked read.
    thread.join(POLL_INTERVAL + 1.0)
    if thread.alive?
      # Last resort: a connect attempt wedged before the read loop.
      thread.kill
      close_active_connection
    end
  end
  # Set authoritatively after the thread is dead so a racing connect
  # call (which also sets "connecting") cannot clobber this value.
  @connection_status = "disconnected"
end