Module: Phlex::Reactive::Streamable

Extended by:
ActiveSupport::Concern
Included in:
Component
Defined in:
lib/phlex/reactive/streamable.rb

Overview

Streamable gives a self-contained Phlex component the ability to render ITSELF as a Turbo Stream and to broadcast itself over a stream. Every streamable component must implement #id returning a stable DOM id — that id is the Turbo Stream target, so you never hand-pick targets.

Class methods (use in controllers):

render turbo_stream: Counter.replace(counter)
render turbo_stream: [Row.append(target: "items", model: @item),
                    Totals.update(@order)]

Broadcast methods (use in models/jobs/actions):

Counter.broadcast_replace_to(counter, model: counter)
Row.broadcast_append_to(@list, target: "items", model: @item)

Convention: the id you set on the root element in view_template must equal what #id returns, so replace/broadcast_replace target it.

NOTE: we intentionally do NOT include Turbo::Streams::ActionHelper — it pulls in ActionView::Helpers::TagHelper, which overrides Phlex's internal tag method and breaks rendering. We use Turbo::Streams::TagBuilder directly instead.

Defined Under Namespace

Classes: ThreadViewContext

Constant Summary collapse

BROADCAST_REFUSED_OPS =

Actor-only ops: a js broadcast rejects a broadcast that carries one. Focus ops steal focus in every subscriber's tab (issue #96); submit (issue #226) would force-submit every subscriber's form; paste_into (issue #228) would read every subscriber's clipboard. They belong to the actor's own reply (reply.js) or gesture (on_client / a reducer's $ops), never a broadcast. Names mirror Phlex::Reactive::JS's verbs.

%w[focus focus_first submit paste_into persist_state persist_clear].freeze
BROADCAST_VERBS =

The broadcast_to verb kwargs (issue #185) → their Turbo stream action. SELF-TARGETING verbs derive the target from the component's #id (require a Streamable payload); CONTAINER verbs need an explicit target: and accept any Phlex component; :js rides the reactive:js custom action.

{
  replace: "replace", update: "update", append: "append",
  prepend: "prepend", remove: "remove", js: "reactive:js"
}.freeze
BROADCAST_SELF_TARGETING = 
%i[replace remove].freeze
BROADCAST_CONTAINER = 
%i[update append prepend].freeze
BROADCAST_MORPHABLE = 
%i[replace update].freeze
REMOVED_BROADCASTS =

Issue #185: the 11 broadcast_*_to / _to_each methods are removed — each raises a guided error printing the broadcast_to rewrite for that verb. (A module constant, not defined inside class_methods, so it's a clean top-level definition the removal loop below reads.)

{
  broadcast_replace_to: "broadcast_to(*keys, replace: model, morph: …)",
  broadcast_update_to: "broadcast_to(*keys, update: model, morph: …)",
  broadcast_append_to: "broadcast_to(*keys, append: model, target: …)",
  broadcast_prepend_to: "broadcast_to(*keys, prepend: model, target: …)",
  broadcast_remove_to: "broadcast_to(*keys, remove: model)",
  broadcast_js_to: "broadcast_to(*keys, js: ops)",
  broadcast_replace_to_each: "broadcast_to(each: keys, replace: model)",
  broadcast_update_to_each: "broadcast_to(each: keys, update: model)",
  broadcast_append_to_each: "broadcast_to(each: keys, append: model, target: …)",
  broadcast_prepend_to_each: "broadcast_to(each: keys, prepend: model, target: …)",
  broadcast_remove_to_each: "broadcast_to(each: keys, remove: model)"
}.freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.broadcast_component(owner, verb, payload, component, keys, morph:, target:, exclude:, visible_to:, effect: nil) ⇒ Object

The ONE broadcast implementation shared by the class-level Streamable.broadcast_to and the module-level Phlex::Reactive.broadcast_to (issue #185). owner is the Streamable class (for the instrument name + its build/render seam); component is the built payload (nil for :js); keys is a list of key-part arrays (one per fan-out key). Renders ONCE (instrumented → render.phlex_reactive) and loops the cheap channel call per key, all wrapped in the broadcast.phlex_reactive event + the pgbus thread-local path (so exclude:/visible_to: reach pgbus and Action Cable no-ops). Self-targeting verbs derive the target from the component's #id and REQUIRE a Streamable payload; container verbs need an explicit target.



121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/phlex/reactive/streamable.rb', line 121

def broadcast_component(owner, verb, payload, component, keys, morph:, target:, exclude:, visible_to:, effect: nil)
  if verb == :js && !effect.nil?
    raise ArgumentError,
      "broadcast_to js: takes no effect: — effects animate element streams (replace/update/" \
      "append/prepend/remove), not client-op dispatches"
  end
  resolved_target = resolve_broadcast_target(verb, component, payload, target)
  # The instrumentation + pgbus thread-locals are owned HERE so the
  # class-level and module-level (plain-component) forms share ONE path
  # and neither is silently un-instrumented (issue #185). The js ops
  # serializer lives HERE too (issue #228): when it was a private method
  # on the includer class, the module-level owner (the Streamable module
  # itself) crashed with NoMethodError instead of refusing — the
  # actor-only gate must be reachable from BOTH broadcast doors.
  component_name = component ? component.class.name : owner.name
  Phlex::Reactive.instrument(
    "broadcast", { component: component_name, stream_action: BROADCAST_VERBS[verb], streamables: keys.size }
  ) do
    with_pgbus_broadcast_opts(exclude:, visible_to:) do
      # A broadcast render NEVER inherits the actor's url_options (issue
      # #232): this call may run inside an action request (where the
      # endpoint threaded the actor's host), but subscribers can be on
      # different hosts — absolute URLs in broadcast-rendered components
      # keep the process defaults, so "host-relative URLs" stays the
      # broadcast contract. Off-request callers pay one nil-write pair.
      html = verb == :js ? nil : Phlex::Reactive.with_url_options(nil) { render_broadcast_html(component) }
      ops_json = verb == :js ? broadcast_js_ops_json(payload) : nil
      keys.each { dispatch_broadcast(verb, it, resolved_target, html, ops_json, morph, effect) }
    end
  end
end

.broadcast_js_ops_json(ops) ⇒ Object

Validate + serialize broadcast ops: reject actor-only ops (focus steals focus in every tab; submit force-submits every subscriber's form; paste_into reads every subscriber's clipboard) and an empty chain (a dead broadcast), then return the JSON wire form. Works on a JS chain (inspect .ops) and a raw array. Lives on the module singleton — the ONE enforcement point both broadcast doors funnel through.



159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/phlex/reactive/streamable.rb', line 159

def broadcast_js_ops_json(ops)
  pairs = ops.is_a?(Phlex::Reactive::JS) ? ops.ops : Array(ops)
  refused = pairs.map { |name, _| name.to_s } & Phlex::Reactive::Streamable::BROADCAST_REFUSED_OPS
  unless refused.empty?
    raise ArgumentError,
      "broadcast_to(js:) refuses actor-only op(s) #{refused.join(", ")} — broadcasting focus " \
      "steals it in every subscriber's tab; broadcasting submit force-submits every " \
      "subscriber's form; broadcasting paste_into reads every subscriber's clipboard. " \
      "These are actor concerns (reply.js / on_client / $ops)."
  end

  Phlex::Reactive::Response.js_ops_json(ops)
end

.broadcast_wire(morph, effect = nil) ⇒ Object

The ONE broadcast attributes compiler (issue #185, extended #215): the broadcast path emits extra attributes via attributes: (it has no method: kwarg) — method="morph" and/or the per-call data-reactive-effect. {} when neither, so the plain call's wire is byte-identical. (Replaces morph_wire.)



261
262
263
264
265
266
267
268
 
# File 'lib/phlex/reactive/streamable.rb', line 261

def broadcast_wire(morph, effect = nil)
  return {} if !morph && effect.nil?

  attrs = {}
  attrs[:method] = "morph" if morph
  attrs[:data] = { reactive_effect: Phlex::Reactive::Effects.wire_value(effect) } unless effect.nil?
  { attributes: attrs }
end

.dispatch_broadcast(verb, key, target, html, ops_json, morph, effect = nil) ⇒ Object

Route ONE key to its Turbo::StreamsChannel call for the verb (issue #185). exclude:/visible_to: are NOT passed here — they ride the pgbus thread-locals set by with_pgbus_broadcast_opts (turbo-rails would swallow them as kwargs). effect (issue #215) rides the same attributes: seam morph does — every broadcast_*_to delegates to broadcast_action_to(attributes:), so the attr lands on the element on every verb.



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/phlex/reactive/streamable.rb', line 236

def dispatch_broadcast(verb, key, target, html, ops_json, morph, effect = nil)
  parts = Array(key)
  wire = broadcast_wire(morph, effect)
  case verb
  when :replace then ::Turbo::StreamsChannel.broadcast_replace_to(*parts, target:, html:, **wire)
  when :update then ::Turbo::StreamsChannel.broadcast_update_to(*parts, target:, html:, **wire)
  when :append then ::Turbo::StreamsChannel.broadcast_append_to(*parts, target:, html:, **wire)
  when :prepend then ::Turbo::StreamsChannel.broadcast_prepend_to(*parts, target:, html:, **wire)
  when :remove then ::Turbo::StreamsChannel.broadcast_remove_to(*parts, target:, **wire)
  when :js
    # Issue #237: the broadcast op stream carries the verbose gate too
    # (same server env for every subscriber of this payload).
    data = { reactive_ops: ops_json }
    data[:reactive_verbose] = "true" if Phlex::Reactive.verbose_errors
    ::Turbo::StreamsChannel.broadcast_action_to(
      *parts, action: "reactive:js", target:, attributes: { data: }, render: false
    )
  end
end

.extract_module_broadcast_verb(verb) ⇒ Object

Split the single verb kwarg out of the module-level broadcast_to's **verb (issue #185) — public counterpart of the class-level extract_broadcast_verb.



272
273
274
275
276
277
278
279
 
# File 'lib/phlex/reactive/streamable.rb', line 272

def extract_module_broadcast_verb(verb)
  unless verb.size == 1 && BROADCAST_VERBS.key?(verb.keys.first)
    raise ArgumentError,
      "broadcast_to needs exactly ONE verb kwarg (#{BROADCAST_VERBS.keys.join("/")}), got #{verb.keys.inspect}"
  end

  verb.first
end

.register(klass) ⇒ Object 



93
94
95
 
# File 'lib/phlex/reactive/streamable.rb', line 93

def register(klass)
  @registry_mutex.synchronize { @registry[klass] = true }
end

.registered_classes ⇒ Object

A point-in-time snapshot of every registered streamable class, taken under the registry lock (the doctor iterates it — issue #106). Returns a plain Array so callers never hold the live WeakMap or the lock while working; a class GC'd later simply won't appear next time. Call Rails.application.eager_load! FIRST if you need every app class loaded — the WeakMap only holds classes that have actually been loaded.



103
104
105
106
107
108
109
 
# File 'lib/phlex/reactive/streamable.rb', line 103

def registered_classes
  @registry_mutex.synchronize do
    classes = []
    @registry.each_key { classes << it }
    classes
  end
end

.render_broadcast_html(component) ⇒ Object

Render a built component to HTML for a broadcast, ALWAYS instrumented (issue #185): a Streamable renders through its own render_component (which fires render.phlex_reactive); a plain component renders through the module render wrapped in the SAME render event, so neither path is silent.



177
178
179
180
181
182
183
184
185
 
# File 'lib/phlex/reactive/streamable.rb', line 177

def render_broadcast_html(component)
  if component.is_a?(Phlex::Reactive::Streamable)
    component.class.render_component(component)
  else
    Phlex::Reactive.instrument(
      "render", { component: component.class.name }
    ) { Phlex::Reactive.render(component) }
  end
end

.reset_all_view_contexts! ⇒ Object

Reset every streamable class's cached view context + builder. Called from the engine's reloader (config.to_prepare) so a reloaded renderer controller is never served from a stale memo. No-op outside Rails.



87
88
89
90
91
 
# File 'lib/phlex/reactive/streamable.rb', line 87

def reset_all_view_contexts!
  @registry_mutex.synchronize do
    @registry.each_key(&:reset_turbo_view_context!)
  end
end

.resolve_broadcast_target(verb, component, payload, target) ⇒ Object

Resolve the DOM target for a broadcast (issue #185): a self-targeting verb (replace/remove) uses the component's #id and REQUIRES a Streamable payload (the #id contract) — a plain component gets a guided error steering to update:; a container verb (update/append/prepend) uses the caller's explicit target: (update self-targets a Streamable when no target: given). :js scopes ops by the caller target (nil → document-scoped).

Raises:

  • (ArgumentError) 


213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
 
# File 'lib/phlex/reactive/streamable.rb', line 213

def resolve_broadcast_target(verb, component, payload, target)
  return target if verb == :js

  streamable = component.is_a?(Phlex::Reactive::Streamable)
  if BROADCAST_SELF_TARGETING.include?(verb) || (verb == :update && target.nil?)
    unless streamable
      raise ArgumentError,
        "broadcast_to #{verb}: needs a Streamable payload (its #id is the target). A plain " \
        "component #{payload.class} has no #id — use update: with an explicit target:."
    end
    return component.id
  end
  raise ArgumentError, "broadcast_to #{verb}: needs a target: (the container element's id)." unless target

  target
end

.with_pgbus_broadcast_opts(exclude:, visible_to:) ⇒ Object

Set the pgbus broadcast thread-locals for the block, gated on pgbus_streams? — the SAME capability-gated path the class-level instrument_broadcast uses (issue #185/#187). Duplicated at module level so the shared broadcast_component owns its transport threading. On Action Cable / old pgbus this is a pure yield.



192
193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/phlex/reactive/streamable.rb', line 192

def with_pgbus_broadcast_opts(exclude:, visible_to:)
  return yield unless Phlex::Reactive.pgbus_streams?

  prev_exclude = Thread.current[:pgbus_broadcast_exclude]
  prev_visible = Thread.current[:pgbus_broadcast_visible_to]
  Thread.current[:pgbus_broadcast_exclude] = exclude
  Thread.current[:pgbus_broadcast_visible_to] = visible_to
  yield
ensure
  if Phlex::Reactive.pgbus_streams?
    Thread.current[:pgbus_broadcast_exclude] = prev_exclude
    Thread.current[:pgbus_broadcast_visible_to] = prev_visible
  end
end

Instance Method Details

#dom_id(record, prefix = nil) ⇒ Object

Render-context-free dom_id, safe to use inside #id. The streamable machinery calls #id BEFORE rendering, so Phlex's render-time dom_id helper would raise HelpersCalledBeforeRenderError. This delegates to ActionView::RecordIdentifier, which works anywhere — so def id = dom_id(@todo) is safe.



627
628
629
 
# File 'lib/phlex/reactive/streamable.rb', line 627

def dom_id(record, prefix = nil)
  ::ActionView::RecordIdentifier.dom_id(record, prefix)
end

#id ⇒ Object

The stable DOM id used as the Turbo Stream target. It MUST match the id set on the component's root element in view_template.

Record-backed components (Component's reactive_record :x) get a default for free: dom_id(record) — the id virtually every such component wrote by hand (issue #81). An explicit def id always wins via normal method lookup. Everything else (state-backed, a bare Streamable) still raises: a class-name default would silently collide the moment two instances render on one page. Two DIFFERENT component classes rendering the SAME record on one page also collide on the default — give one a prefixed id: def id = dom_id(@todo, "rich").

Called on every render/stream build, so the default stays lean: a respond_to? (reactive_record_key is Component API — a bare Streamable keeps raising) plus ivar reads via the memoized reactive_record_ivar.

Raises:

  • (NotImplementedError) 


609
610
611
612
613
614
615
616
617
618
619
620
 
# File 'lib/phlex/reactive/streamable.rb', line 609

def id
  klass = self.class
  if klass.respond_to?(:reactive_record_key) && (record_ivar = klass.reactive_record_ivar) &&
     (record = instance_variable_get(record_ivar))
    return dom_id(record)
  end

  raise NotImplementedError,
    "#{klass} must implement #id (the stable DOM id Turbo Streams target) — add e.g. " \
    "`def id = \"my-thing\"`. Record-backed components (reactive_record :x) get " \
    "`dom_id(record)` as the default automatically."
end

#to_stream_morph ⇒ Object

Issue #185: to_stream_morph is removed — the morph flag is a kwarg now. Guided error → to_stream_replace(morph: true).

Raises:

  • (NoMethodError) 


655
656
657
658
 
# File 'lib/phlex/reactive/streamable.rb', line 655

def to_stream_morph
  raise NoMethodError,
    "to_stream_morph was removed in issue #185 — use to_stream_replace(morph: true)"
end

#to_stream_remove(effect: nil) ⇒ Object

Render THIS instance as a remove stream. The component already knows its own #id, so no record/class reconstruction is needed (works for record- and state-backed components alike). Used by reply.remove.



708
709
710
711
712
713
 
# File 'lib/phlex/reactive/streamable.rb', line 708

def to_stream_remove(effect: nil)
  Phlex::Reactive::Stream.wrap(
    Phlex::Reactive::Effects.annotate(self.class.turbo_stream_builder.remove(id), effect),
    action: "remove", target: id, renders_root: false
  )
end

#to_stream_replace(morph: false, effect: nil) ⇒ Object

Render THIS already-built instance as a replace stream (used by the reactive action endpoint after an action mutated state). Wrapped in a Phlex::Reactive::Stream (issue #114) so the endpoint reads the action / target / token-ness structurally instead of regexing the markup; the wire bytes are byte-identical. morph: true emits <turbo-stream action="replace" method="morph"> (issue #28): Turbo 8's bundled Idiomorph morphs the subtree in place — preserving the focused + caret across the re-render — while still carrying the root's fresh data-reactive-token-value (so the signed token refreshes). Default (morph: false) is the plain outerHTML replace, byte-identical to before. Used by reply.replace / reply.morph. The morph: kwarg replaces the deleted to_stream_morph (issue #185). effect: (issue #215) — the per-call effect override on the actor's own reply stream, same wire as the class builders (nil = untouched).



645
646
647
648
649
650
651
 
# File 'lib/phlex/reactive/streamable.rb', line 645

def to_stream_replace(morph: false, effect: nil)
  builder = self.class.turbo_stream_builder
  html = self.class.render_component(self)
  rendered = morph ? builder.replace(id, html:, method: :morph) : builder.replace(id, html:)
  rendered = Phlex::Reactive::Effects.annotate(rendered, effect)
  Phlex::Reactive::Stream.wrap(rendered, action: "replace", target: id, renders_root: true)
end

#to_stream_token ⇒ Object

Render a TOKEN-ONLY refresh stream (issue #30): a tiny <turbo-stream action="reactive:token"> carrying the component's fresh signed token, with NO rendered body. It lets an action update only PART of a component (its own hand-built streams) while still rolling the signed identity token forward — the client reads the next token from this attribute (#extractToken) and an inert client action writes it onto the root (a pure attribute set, so a focused + caret survive). Unlike to_stream_replace, it does NOT re-render the children, so a live input the user is typing into is never torn down. Used by reply.streams.

The component carries its token via Component#reactive_token; a Streamable that isn't a Component (no token) simply has nothing to refresh — guarded by respond_to? so the primitive stays usable on a bare Streamable.

respond_to? MUST include private methods (the true arg): Component defines reactive_token as PRIVATE, so a plain respond_to?(:reactive_token) is false for every Component and the stream silently carries an EMPTY token — which makes any non-self-rendering reply (reply.streams #30, reply.append / reply.remove #35) add-once-only: the first action works, then the stale (here empty) token is rejected on the next dispatch (cosmos#1939). A bare Streamable has no reactive_token method at all, so it still returns false correctly.



695
696
697
698
699
700
701
702
703
 
# File 'lib/phlex/reactive/streamable.rb', line 695

def to_stream_token
  token = respond_to?(:reactive_token, true) ? reactive_token : nil
  html = %(<turbo-stream action="reactive:token" target="#{ERB::Util.html_escape(id)}" data-reactive-token-value="#{ERB::Util.html_escape(token)}"></turbo-stream>)
  # Both interpolations are ERB::Util.html_escape'd, so .html_safe is a
  # byte-identical relabel. renders_root: true — reactive:token IS a
  # self-render for token purposes (it's in SELF_RENDER_ACTIONS) — unifies
  # the actor's own token stream onto the structural path (issue #114).
  Phlex::Reactive::Stream.wrap(html.html_safe, action: "reactive:token", target: id, renders_root: true)
end

#to_stream_update(morph: false, effect: nil) ⇒ Object

morph: true emits <turbo-stream action="update" method="morph"> (issue #113) so Turbo 8 morphs the inner HTML in place, preserving a focused + caret across a per-field update. Default (morph: false) is the unchanged plain update. Passing method: :morph inline (as #to_stream_morph does) keeps the plain call's wire byte-identical. Used by reply.update.



666
667
668
669
670
671
672
 
# File 'lib/phlex/reactive/streamable.rb', line 666

def to_stream_update(morph: false, effect: nil)
  builder = self.class.turbo_stream_builder
  html = self.class.render_component(self)
  rendered = morph ? builder.update(id, html:, method: :morph) : builder.update(id, html:)
  rendered = Phlex::Reactive::Effects.annotate(rendered, effect)
  Phlex::Reactive::Stream.wrap(rendered, action: "update", target: id, renders_root: true)
end