Class: Pgbus::Web::Streamer::StreamEventDispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/web/streamer/stream_event_dispatcher.rb

Overview

The single-threaded consumer of the shared dispatch_queue. Drains three kinds of messages:

- Listener::WakeMessage(queue_name:) — a NOTIFY fired; read_after
the minimum cursor and fan out to every connection on the stream
(both registered and in-flight connects).

- ConnectMessage(connection:) — a new SSE client connected. Runs
the 5-step race-free replay sequence from design doc §6.5:
  1. ensure_listening on the stream (so future WakeMessages
     deliver to the in-flight buffer)
  2. register an in-flight buffer keyed by connection
  3. read_after(connection.since_id) + enqueue to connection
  4. drain the in-flight buffer into the connection (dedup is
     handled by Connection#enqueue's cursor check)
  5. move the connection from in-flight to the main Registry

- DisconnectMessage(connection:) — unregister and, if the stream
now has zero subscribers, eventually unlisten (lazy GC,
implemented in the Streamer sweep rather than here).

All state ownership lives on this one thread: the registry is thread-safe (Phase 2.1) but the in-flight buffers are local to the dispatcher and accessed only from this thread, so no locks.

Named StreamEventDispatcher (rather than just "Dispatcher") to disambiguate from Pgbus::Process::Dispatcher, which is an unrelated worker-side pool coordinator. See issue #98 item 8.

Defined Under Namespace

Classes: ConnectMessage, DisconnectMessage, PresenceTouchMessage, StreamEnvelope, WriteAckMessage

Constant Summary collapse

WakeMessage = 
Listener::WakeMessage
DEFAULT_READ_LIMIT = 
500

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client:, registry:, listener:, dispatch_queue:, logger: Pgbus.logger, read_limit: DEFAULT_READ_LIMIT, filters: nil, config: nil, stream_counter: nil, presence_provider: nil, pump: nil, ack_queue: nil) ⇒ StreamEventDispatcher

Returns a new instance of StreamEventDispatcher.



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/pgbus/web/streamer/stream_event_dispatcher.rb', line 75

def initialize(client:, registry:, listener:, dispatch_queue:,
               logger: Pgbus.logger, read_limit: DEFAULT_READ_LIMIT,
               filters: nil, config: nil, stream_counter: nil,
               presence_provider: nil, pump: nil, ack_queue: nil)
  @client = client
  @registry = registry
  @listener = listener
  @queue = dispatch_queue
  @logger = logger
  @read_limit = read_limit
  # Off-thread durable fanout (issue #321). When @pump is non-nil
  # (streams_writer_threads > 0), handle_durable_wake hands each
  # registered-connection write to the pump instead of writing inline;
  # the pump reports back an accepted-max on @ack_queue, which the
  # dispatcher drains in apply_acks to advance @scanned_cursor. When
  # nil (default), fanout writes stay inline — the pre-#321 behavior.
  @pump = pump
  @ack_queue = ack_queue
  # Vends a presence handle for a logical stream name. Injected so
  # tests can record join/leave/touch without a DB. Production
  # defaults to the real per-stream Presence via Pgbus.stream.
  @presence_provider = presence_provider || ->(name) { Pgbus.stream(name).presence }
  # Filters default to the process-wide registry so production
  # code picks up whatever was registered at boot. Tests inject
  # a fresh Filters instance to avoid cross-test pollution.
  @filters = filters || Pgbus::Streams.filters
  # Config is injected so the Dispatcher can read
  # `streams_stats_enabled` without reaching into the global
  # Pgbus.configuration at every call site. Tests pass a
  # throwaway config to flip the flag independently of the
  # process-wide setting. Falls back to the global config
  # for production call sites that don't specify one.
  @config = config || Pgbus.configuration
  # Two write deadlines (issue #315 item 3). Fanout writes run serially
  # on this single thread, so a slow client stalls the connections
  # queued behind it — the short fanout deadline bounds that stall.
  # Connect-replay writes run once per subscribe (not in the hot loop),
  # so a fresh client catching up from the archive keeps the full
  # deadline and isn't spuriously evicted.
  @fanout_write_deadline_ms  = @config.streams_fanout_write_deadline_ms
  @connect_write_deadline_ms = @config.streams_write_deadline_ms
  @stream_counter = stream_counter || StreamCounter.new
  # stream_name → Array<[connection, Array<Envelope>]>
  @in_flight = Hash.new { |h, k| h[k] = [] }
  # PGMQ full table name (pgbus_<prefix>_<name>) → logical stream
  # name. Populated on connect so handle_wake can translate
  # Listener::WakeMessage#queue_name (a full table name, because
  # that's what PG NOTIFY channels carry) into the logical name
  # used by Registry and the in-flight buffer.
  @full_to_logical = {}
  # Per-connection "scanned" cursor — the highest msg_id this
  # Dispatcher has examined for a given connection, whether or
  # not it was actually delivered. Needed because an audience
  # filter can drop an entire read_after batch; without a
  # separate scan cursor the dispatcher would re-read the
  # same hidden window forever and starve later public
  # messages. Connection#last_msg_id_sent still drives the
  # client-visible Last-Event-ID; this cursor only feeds
  # minimum_cursor so subsequent read_after calls advance.
  @scanned_cursor = {}
  # @running is a soft hint, not the authoritative stop signal.
  # The :__stop__ sentinel pushed onto @queue is what actually
  # terminates run_loop — even if a torn read of @running ever
  # happened (it cannot under MRI's GVL for a single-word
  # boolean assignment), the sentinel break would still fire.
  @running = false
  @thread = nil
  @ephemeral_seq = 0
end

Instance Attribute Details

#stream_counter ⇒ Object (readonly)

Returns the value of attribute stream_counter.



73
74
75
 
# File 'lib/pgbus/web/streamer/stream_event_dispatcher.rb', line 73

def stream_counter
  @stream_counter
end

Instance Method Details

#start ⇒ Object 



145
146
147
148
149
150
151
 
# File 'lib/pgbus/web/streamer/stream_event_dispatcher.rb', line 145

def start
  return if @running

  @running = true
  @thread = Thread.new { run_loop }
  self
end

#stop ⇒ Object 



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/pgbus/web/streamer/stream_event_dispatcher.rb', line 153

def stop
  return unless @running

  @running = false
  @queue << :__stop__
  if @thread && @thread.join(5).nil?
    # join returned nil → 5s timeout. The thread is still running
    # (probably blocked inside an unresponsive client write or a
    # slow Postgres query). We log and clear the reference rather
    # than calling Thread#kill, which leaves IO state corrupt.
    # The orphaned thread will exit on its own once the blocking
    # call returns and it sees @running == false on the next loop.
    @logger.warn { "[Pgbus::Streamer::StreamEventDispatcher] thread did not terminate within 5s" }
  end
  @thread = nil
  self
end