Module: Pgbus::Streams

Defined in:
lib/pgbus/streams.rb,
lib/pgbus/streams/key.rb,
lib/pgbus/streams/cursor.rb,
lib/pgbus/streams/filters.rb,
lib/pgbus/streams/envelope.rb,
lib/pgbus/streams/presence.rb,
lib/pgbus/streams/renderer.rb,
lib/pgbus/streams/coalescer.rb,
lib/pgbus/streams/streamable.rb,
lib/pgbus/streams/signed_name.rb,
lib/pgbus/streams/pool_trigger.rb,
lib/pgbus/streams/pool_autoscaler.rb,
lib/pgbus/streams/turbo_broadcastable.rb,
lib/pgbus/streams/turbo_stream_override.rb,
lib/pgbus/streams/broadcastable_override.rb,
lib/pgbus/streams/watermark_cache_middleware.rb

Overview

Pgbus::Streams is the SSE-based pub/sub subsystem that replaces turbo_stream_from. The user-facing entrypoint is Pgbus.stream(name), which returns a Pgbus::Streams::Stream providing #broadcast, #current_msg_id, and #read_after.

Defined Under Namespace

Modules: BroadcastableOverride, Cursor, Envelope, Key, Renderer, SignedName, Streamable, TurboBroadcastable, TurboStreamOverride Classes: Coalescer, Filters, PoolAutoscaler, PoolTrigger, Presence, Stream, StreamNameTooLong, WatermarkCacheMiddleware

Constant Summary collapse

DEFAULT_SSE_EVENT =

The default SSE event: name for a broadcast frame. Turbo's StreamObserver consumes frames the client re-dispatches as the message DOM event; the client maps this SSE event name to message. Broadcasts may override it with a typed name (issue #170).

"turbo-stream"

Class Method Summary collapse

Class Method Details

.coalescer ⇒ Object

Process-wide publish-side coalescer for high-frequency broadcasts (issue #171). Lazily built; the flush re-enters the normal broadcast path so a coalesced frame is just a deferred ordinary broadcast.



40
41
42
43
44
45
46
47
48
49
 
# File 'lib/pgbus/streams.rb', line 40

def self.coalescer
  # `target:` is part of the flush signature but unused here — it's only
  # a coalescing key; the target is already encoded in the payload's
  # turbo-stream tag. Absorbed via ** so it isn't a broadcast argument.
  @coalescer ||= Coalescer.new(
    flush: lambda do |stream_name:, payload:, opts:, **|
      Pgbus.stream(stream_name).broadcast(payload, **opts)
    end
  )
end

.filters ⇒ Object

Process-wide registry of server-side audience filter predicates. Register filters at boot time via:

Pgbus::Streams.filters.register(:admin_only) { |user| user.admin? }

See lib/pgbus/streams/filters.rb for the full API.



28
29
30
 
# File 'lib/pgbus/streams.rb', line 28

def self.filters
  @filters ||= Filters.new
end

.install_broadcast_queue!(queue_name) ⇒ Object

turbo-rails' async broadcast helpers (broadcast_*_later_to, the default for broadcasts_to/broadcasts_refreshes) enqueue these three ActiveJobs, which ship with no queue_as and so land on the default queue — where a render+broadcast can wait behind long-running jobs before the browser sees the update (#311). When the operator sets config.streams_broadcast_queue, route them to that dedicated queue so a realtime:-style worker capsule can isolate broadcast latency from job throughput. Called from the engine's turbo_broadcastable initializer. No-op when the queue is nil or turbo-rails is not loaded.



75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/pgbus/streams/turbo_broadcastable.rb', line 75

def self.install_broadcast_queue!(queue_name)
  return if queue_name.nil?

  %w[
    Turbo::Streams::ActionBroadcastJob
    Turbo::Streams::BroadcastJob
    Turbo::Streams::BroadcastStreamJob
  ].each do |const_name|
    next unless Object.const_defined?(const_name)

    Object.const_get(const_name).queue_as(queue_name)
  end
end

.install_turbo_broadcastable_patch! ⇒ Object

Apply the patch to Turbo::StreamsChannel's singleton class. Idempotent: prepending the same module twice is a no-op. Called from Pgbus::Engine's initializer when Turbo is detected.



59
60
61
62
63
64
 
# File 'lib/pgbus/streams/turbo_broadcastable.rb', line 59

def self.install_turbo_broadcastable_patch!
  return unless defined?(::Turbo::StreamsChannel)
  return if ::Turbo::StreamsChannel.singleton_class.include?(TurboBroadcastable)

  ::Turbo::StreamsChannel.singleton_class.prepend(TurboBroadcastable)
end

.install_turbo_stream_override! ⇒ Object

Apply the patch to Turbo::StreamsHelper. Idempotent: prepending the same module twice is a no-op. Called from Pgbus::Engine's initializer when both turbo-rails and pgbus streams are enabled.



43
44
45
46
47
48
 
# File 'lib/pgbus/streams/turbo_stream_override.rb', line 43

def self.install_turbo_stream_override!
  return unless defined?(::Turbo::StreamsHelper)
  return if ::Turbo::StreamsHelper.ancestors.include?(TurboStreamOverride)

  ::Turbo::StreamsHelper.prepend(TurboStreamOverride)
end

.reset_coalescer! ⇒ Object

Clears the coalescer. Used by tests; not intended for runtime.



52
53
54
 
# File 'lib/pgbus/streams.rb', line 52

def self.reset_coalescer!
  @coalescer = nil
end

.reset_filters! ⇒ Object

Clears the filters registry. Used by tests; not intended for runtime.



33
34
35
 
# File 'lib/pgbus/streams.rb', line 33

def self.reset_filters!
  @filters = nil
end