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 =

Focus ops are actor-only (they steal focus): a js broadcast rejects a broadcast that carries one. Names mirror Phlex::Reactive::JS's focus verbs.

%w[focus focus_first].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.



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/phlex/reactive/streamable.rb', line 117

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)
  # broadcast_js_ops_json is a private class method on the owner Streamable
  # class (reached via send); 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).
  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 ? owner.send(:broadcast_js_ops_json, payload) : nil
      keys.each { dispatch_broadcast(verb, it, resolved_target, html, ops_json, morph, effect) }
    end
  end
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.)



224
225
226
227
228
229
230
231
 
# File 'lib/phlex/reactive/streamable.rb', line 224

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.



203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
 
# File 'lib/phlex/reactive/streamable.rb', line 203

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.



235
236
237
238
239
240
241
242
 
# File 'lib/phlex/reactive/streamable.rb', line 235

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 



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

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.



99
100
101
102
103
104
105
 
# File 'lib/phlex/reactive/streamable.rb', line 99

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.



144
145
146
147
148
149
150
151
152
 
# File 'lib/phlex/reactive/streamable.rb', line 144

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.



83
84
85
86
87
 
# File 'lib/phlex/reactive/streamable.rb', line 83

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) 


180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/phlex/reactive/streamable.rb', line 180

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.



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

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.



606
607
608
 
# File 'lib/phlex/reactive/streamable.rb', line 606

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) 


588
589
590
591
592
593
594
595
596
597
598
599
 
# File 'lib/phlex/reactive/streamable.rb', line 588

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) 


634
635
636
637
 
# File 'lib/phlex/reactive/streamable.rb', line 634

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.



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

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



624
625
626
627
628
629
630
 
# File 'lib/phlex/reactive/streamable.rb', line 624

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.



674
675
676
677
678
679
680
681
682
 
# File 'lib/phlex/reactive/streamable.rb', line 674

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.



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

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