Class: Insika::EventStream

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

Overview

In-process pub/sub. The stream is concurrent with the turn: a slow observer NEVER delays execution — each subscriber has its own queue and emit only enqueues. No mutex: one reactor, cooperative fibers — a plain Array is enough.

Defined Under Namespace

Classes: Subscription

Instance Method Summary collapse

Constructor Details

#initialize ⇒ EventStream

Returns a new instance of EventStream.



99
100
101
 
# File 'lib/insika/event_stream.rb', line 99

def initialize
  @subscriptions = []
end

Instance Method Details

#emit(event) ⇒ Object

NEVER raises: an observer's exception is isolated — a broken observer does not bring down the turn. Synchronous and cheap.

Iterates over a SNAPSHOT (dup): a subscription's cap may close it DURING the push (overflow -> close -> on_close removes from the array); mutating the array in the middle of a plain Array#each would skip the next subscriber.



110
111
112
113
114
115
116
117
 
# File 'lib/insika/event_stream.rb', line 110

def emit(event)
  @subscriptions.dup.each do |sub|
    sub.push(event) if sub.matches?(event)
  rescue StandardError
    # a broken observer does not bring down the turn; nothing to propagate
  end
  nil
end

#subscribe(task_id: nil, session_id: nil, tenant: nil, types: nil) ⇒ Object

nil/nil = all events. Returns the Subscription (the caller iterates with #each on its own fiber). tenant: scopes the stream to one tenant's events (WS1) — fail-closed, see Subscription#matches?. types: (nil = any) filters by event type so a subscriber's queue only ever holds what its consumer answers (WS6).



124
125
126
127
128
129
130
 
# File 'lib/insika/event_stream.rb', line 124

def subscribe(task_id: nil, session_id: nil, tenant: nil, types: nil)
  sub = Subscription.new(task_id: task_id, session_id: session_id, tenant: tenant,
                         types: types,
                         on_close: ->(s) { @subscriptions.delete(s) })
  @subscriptions << sub
  sub
end