class OpenUI::Rails::Stream::Broadcaster

Binds the ordered {Bridge} to a transport and broadcasts every envelope as it is produced.

The transport is a seam: any object that responds to ‘#broadcast(envelope)` works, and {Transports} ships wrappers for Action Cable, Turbo Streams and an in-memory collector. Delivery is best effort: Action Cable and Turbo Streams do not promise order or delivery. The client applies sequence numbers and recovers from an authoritative snapshot on a gap (see the Stimulus controller).

broadcaster = Broadcaster.new(
  transport: Transports::ActionCable.new("openui:surface:42"),
  message_id: "msg-1",
  library_version: OpenUI::Rails.configuration.library_fingerprint
)
chat.complete { |chunk| broadcaster.push(chunk.content) }
broadcaster.finish

Attributes

library_version[R]

message_id[R]

transport[R]

Public Class Methods

broadcast(source, **options) click to toggle source

Broadcast every envelope produced by an enumerable source of chunks.

@param source [#each] yields String chunks @param options [Hash] arguments forwarded to the constructor

@return [Broadcaster]

# File lib/openui/rails/stream/broadcaster.rb, line 32
def self.broadcast(source, **options)
  broadcaster = new(**options)
  source.each { |chunk| broadcaster.push(chunk) }
  broadcaster.finish
end

new(transport:, message_id:, library_version: nil, configuration: OpenUI::Rails.configuration) click to toggle source

Bind a bridge writer to a transport that responds to ‘broadcast`.

@param transport [#broadcast]

@param message_id [String]

@param library_version [String, nil]

@param configuration [Configuration]

@return [Broadcaster]

# File lib/openui/rails/stream/broadcaster.rb, line 49
def initialize(transport:, message_id:, library_version: nil, configuration: OpenUI::Rails.configuration)
  raise ConfigurationError, "a transport is required" if transport.nil?
  unless transport.respond_to?(:broadcast)
    raise ConfigurationError, "transport must respond to #broadcast"
  end

  @transport = transport
  @message_id = message_id
  @library_version = library_version
  @configuration = configuration
  @writer = Bridge.new(
    message_id: message_id,
    library_version: library_version,
    configuration: configuration
  ).writer
end

Public Instance Methods

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

Broadcast a structured, PII-free error envelope and stop.

@param error [Exception]

@param code [String]

@return [self]

# File lib/openui/rails/stream/broadcaster.rb, line 91
def fail(error, code: "stream_failed")
  envelope = @writer.error_for(error, code: code)
  @transport.broadcast(envelope) if envelope
  self
end

finish() click to toggle source

Broadcast the final delta (if any) and the completion marker.

@return [self]

# File lib/openui/rails/stream/broadcaster.rb, line 79
def finish
  @writer.finish.each { |envelope| @transport.broadcast(envelope) }
  self
end

push(chunk) click to toggle source

Push a chunk; broadcasts any envelopes the writer flushes.

@param chunk [String, nil]

@return [self]

# File lib/openui/rails/stream/broadcaster.rb, line 71
def push(chunk)
  @writer.push(chunk).each { |envelope| @transport.broadcast(envelope) }
  self
end

sequence(= @writer.sequence) click to toggle source

The next sequence number the writer will assign.

# File lib/openui/rails/stream/broadcaster.rb, line 99
  def sequence = @writer.sequence

  # The underlying writer, for callers that need finer control.
  attr_reader :writer
end