Class: Mistri::SSE

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

Overview

An incremental Server-Sent Events decoder. Feed it raw socket fragments in any chunking; it buffers partial lines across fragments and yields each complete "data:" record as a parsed Hash.

The decode is deliberately tolerant of what the provider APIs actually send: one single-line JSON object per "data:" record. "event:", "id:", comment, and blank lines are ignored (the event name is duplicated inside the data payload), OpenAI's "[DONE]" sentinel is dropped, and a record that fails to parse is skipped rather than killing a live stream. Only the partial line is bounded; the complete stream is not.

Instance Method Summary collapse

Constructor Details

#initialize(max_record_bytes: DEFAULT_MAX_RECORD_BYTES) ⇒ SSE

Returns a new instance of SSE.



20
21
22
23
24
25
26
27
 
# File 'lib/mistri/sse.rb', line 20

def initialize(max_record_bytes: DEFAULT_MAX_RECORD_BYTES)
  unless max_record_bytes.is_a?(Integer) && max_record_bytes.positive?
    raise ConfigurationError, "max_record_bytes: must be a positive integer"
  end

  @max_record_bytes = max_record_bytes
  @buffer = +""
end

Instance Method Details

#feed(fragment) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/mistri/sse.rb', line 29

def feed(fragment, &)
  scan = fragment.encoding == Encoding::BINARY ? fragment : fragment.b
  offset = 0
  while (newline = scan.index("\n", offset))
    append(fragment, offset, newline - offset)
    line = @buffer
    @buffer = +""
    line.chomp!
    decode(line, &)
    offset = newline + 1
  end
  append(fragment, offset, scan.bytesize - offset)
  nil
end

#finish ⇒ Object

Flush a trailing record that arrived without a final newline.



45
46
47
48
49
50
51
52
53
 
# File 'lib/mistri/sse.rb', line 45

def finish(&)
  unless @buffer.empty?
    line = @buffer
    @buffer = +""
    line.chomp!
    decode(line, &)
  end
  nil
end