Class: Mistri::Sinks::Coalesced

Inherits:
Object
  • Object
show all
Defined in:
lib/mistri/sinks/coalesced.rb

Overview

Wraps any sink and merges bursts of streaming deltas, so a transport broadcasts at UI speed instead of token speed. Deltas buffer per content block per origin and flush merged when the interval elapses or any other event arrives, so ordering is preserved and a turn always ends flushed. Safe to share across threads: a background worker's events serialize with the parent's instead of racing the buffer.

sink = Mistri::Sinks::Coalesced.new(
Mistri::Sinks::ActionCable.new("agent_1"), interval: 0.1,
)

Constant Summary collapse

DELTAS = 
%i[text_delta thinking_delta toolcall_delta].freeze

Instance Method Summary collapse

Constructor Details

#initialize(sink, interval: 0.05) ⇒ Coalesced

Returns a new instance of Coalesced.



18
19
20
21
22
23
24
 
# File 'lib/mistri/sinks/coalesced.rb', line 18

def initialize(sink, interval: 0.05)
  @sink = sink
  @interval = interval
  @buffer = nil
  @flushed_at = now
  @mutex = Mutex.new
end

Instance Method Details

#call(event) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/mistri/sinks/coalesced.rb', line 26

def call(event)
  @mutex.synchronize do
    if DELTAS.include?(event.type)
      merge(event)
      flush if now - @flushed_at >= @interval
    else
      flush
      @sink.call(event)
    end
  end
end

#to_proc ⇒ Object 



38
 
# File 'lib/mistri/sinks/coalesced.rb', line 38

def to_proc = method(:call).to_proc