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, tenant: nil, types: nil, on_close: nil) ⇒ Subscription

Returns a new instance of Subscription.



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

def initialize(task_id: nil, session_id: nil, tenant: nil, types: nil, on_close: nil)
  @task_id = task_id
  @session_id = session_id
  @tenant = tenant
  @types = types
  @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.



37
38
39
40
 
# File 'lib/insika/event_stream.rb', line 37

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).



90
91
92
93
94
95
96
 
# File 'lib/insika/event_stream.rb', line 90

def close
  return if @closed

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

#each ⇒ Object

Blocks the CONSUMER's fiber until #close.



82
83
84
85
86
 
# File 'lib/insika/event_stream.rb', line 82

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.

A TENANT-scoped subscription is FAIL-CLOSED on the meta's tenant (WS1): an event that does not carry the tenant (control events, ignored turns) matches NO tenant subscription. The tenant only ever sees its own.

types: (nil = any) keeps a subscriber's queue to the events it answers — an alert consumer must not sit behind a full-traffic stream's 1000-cap (WS6), and a filtered queue is the cheapest way to keep it there.

Returns:

  • (Boolean) 


52
53
54
55
56
57
58
59
60
 
# File 'lib/insika/event_stream.rb', line 52

def matches?(event)
  meta = event.meta || {}
  owned = @tenant.nil? || meta[:tenant] == @tenant

  owned &&
    (@types.nil? || @types.include?(event.type)) &&
    (@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).



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/insika/event_stream.rb', line 65

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