Class: Insika::EventStream::Subscription

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

Overview

One subscription = one queue. The consumer blocks on each (its own fiber), never the emitter.

Constant Summary collapse

MAX_QUEUED =

Cap on events queued per subscriber. A slow consumer piles up in its OWN queue; on overflow, the subscription closes with a local :error event — the turn never waits on transport.

1000

Instance Method Summary collapse

Constructor Details

#initialize(task_id: nil, session_id: nil, on_close: nil) ⇒ Subscription

Returns a new instance of Subscription.



23
24
25
26
27
28
 
# File 'lib/insika/event_stream.rb', line 23

def initialize(task_id: nil, session_id: nil, on_close: nil)
  @task_id = task_id
  @session_id = session_id
  @on_close = on_close
  @queue = Async::Queue.new
end

Instance Method Details

#bind(task_id:) ⇒ Object

Binds the subscription to a task_id AFTER creation: the transport subscribes before the dispatch — when the task_id does not yet exist — and binds it as soon as the handler returns. This keeps the per-task cap honest (only events for THIS task enter the queue) and makes the overflow :error carry the correct task_id. Returns self to chain.



35
36
37
38
 
# File 'lib/insika/event_stream.rb', line 35

def bind(task_id:)
  @task_id = task_id
  self
end

#close ⇒ Object

Idempotent: a second CLOSED is harmless (the each stops at the first). @on_close fires only once (avoids removing the subscription twice).



76
77
78
79
80
81
82
 
# File 'lib/insika/event_stream.rb', line 76

def close
  return if @closed

  @closed = true
  @queue.enqueue(CLOSED)
  @on_close&.call(self)
end

#each ⇒ Object

Blocks the CONSUMER's fiber until #close.



68
69
70
71
72
 
# File 'lib/insika/event_stream.rb', line 68

def each
  while (event = @queue.dequeue) != CLOSED
    yield event
  end
end

#matches?(event) ⇒ Boolean

Meta filter: nil = matches any value. Events with no task_id in meta (e.g. :session_created) reach only subscribers with no task filter.

Returns:

  • (Boolean) 


42
43
44
45
46
 
# File 'lib/insika/event_stream.rb', line 42

def matches?(event)
  meta = event.meta || {}
  (@task_id.nil? || meta[:task_id] == @task_id) &&
    (@session_id.nil? || meta[:session_id] == @session_id)
end

#push(event) ⇒ Object

Enqueues without EVER blocking. The real queue depth is @queue.size (not a separate counter — avoids drift). On reaching the cap, it enqueues a local :error and closes; pushes after close are ignored (@closed).



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/insika/event_stream.rb', line 51

def push(event)
  return if @closed

  if @queue.size >= MAX_QUEUED
    @queue.enqueue(Insika::Event.new(
                     type: :error,
                     data: { message: "subscription overflow" },
                     meta: { task_id: @task_id, session_id: @session_id }
                   ))
    close
    return
  end

  @queue.enqueue(event)
end