Module: Pgbus::Client::NotifyStream

Included in:
Pgbus::Client
Defined in:
lib/pgbus/client/notify_stream.rb

Overview

Fire-and-forget PG NOTIFY for ephemeral stream broadcasts. No PGMQ queue is created — the payload travels via the Postgres NOTIFY channel only, matching the channel naming convention that PGMQ's trigger uses:

pgmq.q_<full_queue_name>.INSERT

Subscribers already LISTEN on this channel via the Streamer's Listener. When a subscriber is connected, the StreamEventDispatcher receives the NOTIFY and fans out the payload. When no subscriber is connected, the NOTIFY is silently discarded by Postgres — no queue, no storage, no orphan tables.

The payload is JSON-serialized into the NOTIFY's optional payload parameter. Postgres caps NOTIFY payloads at < 8000 bytes; oversized payloads raise a typed Pgbus::Streams::PayloadTooLarge here, at the call site, instead of surfacing as a misleading PGMQ::Errors::ConnectionError ("payload string too long") from deep inside the driver (issue #391). Callers needing large payloads should use durable mode (which inserts into PGMQ).

Constant Summary collapse

NOTIFY_PAYLOAD_LIMIT_BYTES =

PostgreSQL rejects NOTIFY payloads of 8000 bytes or more ("payload string too long"), so 7999 is the largest deliverable payload.

7999

Instance Method Summary collapse

Instance Method Details

#notify_stream(stream_name, payload) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/pgbus/client/notify_stream.rb', line 29

def notify_stream(stream_name, payload)
  full_name = config.queue_name(stream_name)
  sanitized = QueueNameValidator.sanitize!(full_name)
  channel = "pgmq.q_#{sanitized}.INSERT"
  json = payload.is_a?(String) ? payload : JSON.generate(payload)
  validate_notify_payload_size!(stream_name, json)

  Instrumentation.instrument("pgbus.stream.notify", stream: stream_name, bytes: json.bytesize) do
    with_stale_connection_retry do
      synchronized do
        @pgmq.__send__(:with_connection) do |conn|
          conn.exec_params("SELECT pg_notify($1, $2)", [channel, json])
        end
      end
    end
  end
end