class OpenUI::Rails::Stream::Bridge::Writer

Accumulates pushed chunks and emits envelopes. Stateful across calls.

Attributes

sequence[R]

Public Class Methods

new(message_id:, library_version: nil, pending_byte_limit:, stream_byte_limit:, on_error: nil) click to toggle source

Create a writer that buffers chunks up to the pending byte limit.

@param message_id [String]

@param library_version [String, nil]

@param pending_byte_limit [Integer]

@param stream_byte_limit [Integer]

@param on_error [#call, nil]

@return [Writer]

# File lib/openui/rails/stream/bridge.rb, line 95
def initialize(message_id:, library_version: nil, pending_byte_limit:, stream_byte_limit:, on_error: nil)
  Envelope.completed(message_id: message_id, sequence: 0, library_version: library_version)
  unless pending_byte_limit.is_a?(Integer) && stream_byte_limit.is_a?(Integer) &&
      pending_byte_limit > 0 && stream_byte_limit >= pending_byte_limit
    raise ConfigurationError, "stream limits must be positive integers with pending <= total"
  end
  @message_id = message_id
  @library_version = library_version
  @pending_byte_limit = pending_byte_limit
  @stream_byte_limit = stream_byte_limit
  @on_error = on_error
  @sequence = 0
  @total_bytes = 0
  @pending = +""
  @terminated = false
end

Public Instance Methods

error_for(error, code: "stream_failed") click to toggle source

Structured, PII-free error envelope for a provider/consumer failure.

@param error [Exception]

@param code [String] a supported stream error code

@return [Envelope, nil] nil when the writer already terminated

# File lib/openui/rails/stream/bridge.rb, line 157
def error_for(error, code: "stream_failed")
  return nil if @terminated

  @on_error&.call(error, message_id: @message_id)

  message = "stream failed (#{error.class})"
  message = "stream cancelled" if code == "stream_cancelled"
  error_envelope(code, message)
end

finish() click to toggle source

Flush pending text and emit the completion envelope once.

@return [Array<Envelope>] the final delta, if any, and completion

# File lib/openui/rails/stream/bridge.rb, line 137
def finish
  return [] if @terminated

  envelopes = []
  envelopes << flush_delta unless @pending.empty?
  envelopes << Envelope.completed(
    message_id: @message_id, sequence: @sequence, library_version: @library_version
  )
  @sequence += 1
  @terminated = true
  envelopes
end

push(chunk) click to toggle source

Append a chunk and return a delta once the pending threshold is reached. A chunk that exceeds the total stream limit returns an error envelope.

@param chunk [String, nil]

@return [Array<Envelope>] zero or more envelopes to emit

# File lib/openui/rails/stream/bridge.rb, line 118
def push(chunk)
  return [] if @terminated || chunk.nil?
  raise StreamError, "stream chunks must be strings" unless chunk.is_a?(String)
  return [] if chunk.empty?

  @total_bytes += chunk.bytesize
  if @total_bytes > @stream_byte_limit
    return [ error_envelope("stream_too_large", "stream exceeded #{@stream_byte_limit} bytes") ]
  end

  @pending << chunk
  return [] if @pending.bytesize < @pending_byte_limit

  [ flush_delta ]
end