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.



85
86
87
 
# File 'lib/insika/event_stream.rb', line 85

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.



96
97
98
99
100
101
102
103
 
# File 'lib/insika/event_stream.rb', line 96

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) ⇒ Object

nil/nil = all events. Returns the Subscription (the caller iterates with #each on its own fiber).



107
108
109
110
111
112
 
# File 'lib/insika/event_stream.rb', line 107

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