Class: RubyPi::LLM::SSEParser

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_pi/llm/sse_parser.rb

Overview

Incremental, bounded Server-Sent Events decoder.

Faraday delivers arbitrary byte chunks, including chunks split inside a UTF-8 character or line ending. This decoder keeps bytes in BINARY form until a complete SSE event is available, then validates UTF-8 and yields the joined value of its data fields.

Constant Summary collapse

DEFAULT_MAX_EVENT_BYTES = 
1_048_576

Instance Method Summary collapse

Constructor Details

#initialize(provider:, max_event_bytes: DEFAULT_MAX_EVENT_BYTES) ⇒ SSEParser

Returns a new instance of SSEParser.



16
17
18
19
20
21
22
 
# File 'lib/ruby_pi/llm/sse_parser.rb', line 16

def initialize(provider:, max_event_bytes: DEFAULT_MAX_EVENT_BYTES)
  @provider = provider
  @max_event_bytes = max_event_bytes
  @buffer = (+"").force_encoding(Encoding::BINARY)
  @data_lines = []
  @event_bytes = 0
end

Instance Method Details

#feed(chunk, &block) ⇒ Object 



24
25
26
27
28
 
# File 'lib/ruby_pi/llm/sse_parser.rb', line 24

def feed(chunk, &block)
  @buffer << chunk.to_s.b
  extract_lines(finishing: false, &block)
  enforce_size!(@buffer.bytesize + @event_bytes)
end

#finish(&block) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/ruby_pi/llm/sse_parser.rb', line 30

def finish(&block)
  extract_lines(finishing: true, &block)

  unless @buffer.empty?
    line = @buffer.slice!(0, @buffer.bytesize)
    process_line(line, &block)
  end

  dispatch(&block)
end

#parse_json(payload) ⇒ Object 



41
42
43
44
45
46
47
48
 
# File 'lib/ruby_pi/llm/sse_parser.rb', line 41

def parse_json(payload)
  JSON.parse(payload)
rescue JSON::ParserError => e
  raise RubyPi::StreamingProtocolError.new(
    "#{@provider} stream contained malformed JSON: #{e.message}",
    status_code: nil
  )
end