Class: MaquinaStream::Broadcaster

Inherits:
Object
  • Object
show all
Defined in:
lib/maquina_stream/broadcaster.rb

Overview

Turns a growing buffer into a stream of small patches.

broadcaster = MaquinaStream::Broadcaster.new(message)
model.stream { |token| broadcaster.append(token) }
broadcaster.seal!                 # final frame, always

It remembers what the browser already has — block id to digest — and sends only what moved. A block that has not changed is never re-sent, which is the difference between bandwidth tracking drift and bandwidth tracking message length.

Transport is a seam. It defaults to Turbo Streams over whatever cable the host configured, and a test can hand it a recorder instead; that recorder is how the bandwidth budget is asserted rather than estimated.

One broadcaster per stream, held for the life of that stream: what the browser already has lives in the instance, so a fresh broadcaster mid-stream re-sends every block. It is not thread-safe; drive one stream from one place.

Deltas do not converge on their own, and are not meant to. Only the open tail is patched; a block that changes after it stops being the tail is fixed by the repair path. Correctness lives there.

Defined Under Namespace

Classes: TurboTransport

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(record, config: MaquinaStream.config, transport: nil) ⇒ Broadcaster

Builds a broadcaster over one record.

transport: defaults to TurboTransport. It is a seam: anything answering call(record:, frame:, config:) will do, which is how the bandwidth budget is asserted against a recorder rather than estimated.



48
49
50
51
52
53
54
55
 
# File 'lib/maquina_stream/broadcaster.rb', line 48

def initialize(record, config: MaquinaStream.config, transport: nil)
  @record = record
  @config = config
  @transport = transport || TurboTransport.new
  @known = {}
  @tail_id = nil
  @last_flush = nil
end

Instance Attribute Details

#config ⇒ Object (readonly)

The Configuration this broadcaster reads — frame_budget_ms and seal_lag in particular.



37
38
39
 
# File 'lib/maquina_stream/broadcaster.rb', line 37

def config
  @config
end

#record ⇒ Object (readonly)

The host record being streamed. Must satisfy MaquinaStream::Streamable.



33
34
35
 
# File 'lib/maquina_stream/broadcaster.rb', line 33

def record
  @record
end

#transport ⇒ Object (readonly)

The object frames are emitted through. Responds to call(record:, frame:, config:).



41
42
43
 
# File 'lib/maquina_stream/broadcaster.rb', line 41

def transport
  @transport
end

Instance Method Details

#append(text, now: monotonic_ms) ⇒ Object

Appends text to the record's buffer and broadcasts if the frame budget has elapsed. Returns the Frame that went out, or nil when this append was coalesced into the next one.

This is the method a streaming loop calls, once per token or per chunk.

Host owns persistence: it appends to its own column, and only then is there anything to broadcast.



65
66
67
68
 
# File 'lib/maquina_stream/broadcaster.rb', line 65

def append(text, now: monotonic_ms)
  record.maquina_stream_append(text)
  broadcast(now: now)
end

#broadcast(now: monotonic_ms) ⇒ Object

Broadcasts a frame if the budget has elapsed, without appending anything. Returns the Frame or nil.

Useful when the buffer moved by some other route — a host that writes to the column itself and wants the browser told about it.

Coalescing happens BEFORE the render, not after it. Building a frame means rendering the whole buffer, so doing that per token and then throwing the result away is how a stream becomes quadratic in message length.

Skipping a frame costs nothing: the next one is computed against what the browser actually has, so it carries the accumulated difference.



95
96
97
98
99
 
# File 'lib/maquina_stream/broadcaster.rb', line 95

def broadcast(now: monotonic_ms)
  return nil unless due?(now)

  emit(now: now)
end

#document ⇒ Object

A Document over the record's current buffer, rendered in the mode its status implies. Rebuilt on every call, because the buffer moves.

A sealed message renders in static mode, which is what takes the caret off the last block.



106
107
108
109
110
111
112
113
 
# File 'lib/maquina_stream/broadcaster.rb', line 106

def document
  Document.new(
    record.maquina_stream_buffer,
    config: config,
    sid: record.maquina_stream_id,
    mode: record.maquina_stream_open? ? :streaming : :static
  )
end

#seal!(status: :complete) ⇒ Object

Seals the record and emits the final frame. Returns that Frame.

status: is one of Streamable::SEAL_STATUSES. Call this exactly once, including when a stream failed: an errored message still has text worth keeping, and the client has no other way to learn the stream is over.

The final frame is never coalesced and never skipped. Every intra-stream drift becomes cosmetic and self-correcting because of this one.



78
79
80
81
 
# File 'lib/maquina_stream/broadcaster.rb', line 78

def seal!(status: :complete)
  record.maquina_stream_seal!(status: status)
  emit(now: monotonic_ms, final: true)
end