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].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
 
# 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
      html = verb == :js ? 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.



153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/phlex/reactive/streamable.rb', line 153

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.)



251
252
253
254
255
256
257
258
 
# File 'lib/phlex/reactive/streamable.rb', line 251

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.



230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'lib/phlex/reactive/streamable.rb', line 230

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
    ::Turbo::StreamsChannel.broadcast_action_to(
      *parts, action: "reactive:js", target:, attributes: { data: { reactive_ops: ops_json } }, 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.



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

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.



171
172
173
174
175
176
177
178
179
 
# File 'lib/phlex/reactive/streamable.rb', line 171

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) 


207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/phlex/reactive/streamable.rb', line 207

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.



186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/phlex/reactive/streamable.rb', line 186

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.



618
619
620
 
# File 'lib/phlex/reactive/streamable.rb', line 618

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) 


600
601
602
603
604
605
606
607
608
609
610
611
 
# File 'lib/phlex/reactive/streamable.rb', line 600

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) 


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

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.



699
700
701
702
703
704
 
# File 'lib/phlex/reactive/streamable.rb', line 699

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).



636
637
638
639
640
641
642
 
# File 'lib/phlex/reactive/streamable.rb', line 636

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.



686
687
688
689
690
691
692
693
694
 
# File 'lib/phlex/reactive/streamable.rb', line 686

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.



657
658
659
660
661
662
663
 
# File 'lib/phlex/reactive/streamable.rb', line 657

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