Class: RubyLLM::BedrockInvoke::StreamState

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_llm/bedrock_invoke/stream_state.rb

Overview

Per-attempt streaming state: accumulator, block collector, and event-stream decoder.

RubyLLM's Faraday connection installs retry middleware that re-issues POSTs on retryable failures (throttling raised mid-stream, connection drops). The on_data callback closes over this holder; resetting it at the first chunk of each physical attempt prevents a retried stream from appending to a half-filled accumulator or feeding a decoder stuck on a partial frame. Detection: Faraday reports cumulative received bytes, which equal the chunk size exactly when an attempt starts fresh.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(event_stream_owner) ⇒ StreamState

Returns a new instance of StreamState.



18
19
20
21
 
# File 'lib/ruby_llm/bedrock_invoke/stream_state.rb', line 18

def initialize(event_stream_owner)
  @owner = event_stream_owner
  reset!
end

Instance Attribute Details

#accumulator ⇒ Object (readonly)

Returns the value of attribute accumulator.



16
17
18
 
# File 'lib/ruby_llm/bedrock_invoke/stream_state.rb', line 16

def accumulator
  @accumulator
end

#collector ⇒ Object (readonly)

Returns the value of attribute collector.



16
17
18
 
# File 'lib/ruby_llm/bedrock_invoke/stream_state.rb', line 16

def collector
  @collector
end

#decoder ⇒ Object (readonly)

Returns the value of attribute decoder.



16
17
18
 
# File 'lib/ruby_llm/bedrock_invoke/stream_state.rb', line 16

def decoder
  @decoder
end

Instance Method Details

#reset! ⇒ Object 



23
24
25
26
27
 
# File 'lib/ruby_llm/bedrock_invoke/stream_state.rb', line 23

def reset!
  @accumulator = RubyLLM::StreamAccumulator.new
  @collector = BlockCollector.new
  @decoder = @owner.event_stream_decoder
end

#reset_if_new_attempt!(chunk, received_bytes) ⇒ Object 



29
30
31
 
# File 'lib/ruby_llm/bedrock_invoke/stream_state.rb', line 29

def reset_if_new_attempt!(chunk, received_bytes)
  reset! if received_bytes && received_bytes == chunk.bytesize
end