Module: Pgbus::Streams::Envelope

Defined in:
lib/pgbus/streams/envelope.rb

Overview

Encodes Server-Sent Events frames per https://html.spec.whatwg.org/multipage/server-sent-events.html.

Pgbus uses three frame types:

- `message(id:, event:, data:)` — a real broadcast (carries an `id:` so the client
can resume via `Last-Event-ID` on reconnect)
- `comment(text)` — a heartbeat or sentinel that the SSE parser ignores
- `retry_directive(ms)` — tells `EventSource` how long to wait before reconnecting

All frames end with \n\n (the SSE event terminator). A multiline payload is framed as consecutive data: lines (issue #392) — the client rejoins them with \n, so delivery is lossless. \r\n and lone \r are also SSE line terminators, so they become data: line breaks too (rejoined as \n; SSE cannot represent a raw \r). Every payload line carries the data: prefix, so a crafted payload cannot inject forged id:/event: fields. Single-line fields (event:, comments) still strip newlines — there a \r/\n would terminate the field early and permit SSE field injection.

Constant Summary collapse

NEWLINES = 
/[\r\n]+/
DATA_LINE_BREAK = 
/\r\n|\r|\n/
RESPONSE_HEADERS = 
"HTTP/1.1 200 OK\r\n" \
"content-type: text/event-stream\r\n" \
"cache-control: no-cache, no-transform\r\n" \
"x-accel-buffering: no\r\n" \
"connection: keep-alive\r\n" \
"\r\n"

Class Method Summary collapse

Class Method Details

.comment(text) ⇒ Object 



47
48
49
 
# File 'lib/pgbus/streams/envelope.rb', line 47

def self.comment(text)
  ": #{strip_newlines(text.to_s)}\n\n"
end

.connected(id:) ⇒ Object

Emits a pgbus:connected frame carrying the server-minted connection id as JSON. Sent once, right after the open handshake, so the page can read its own connection id and send it back as X-Pgbus-Connection on action requests (actor-echo suppression, issue #165). Deliberately omits an id: line: this is connection metadata, not a broadcast, and giving it a cursor id would corrupt the client's Last-Event-ID replay position on reconnect.

Raises:

  • (ArgumentError) 


58
59
60
61
62
 
# File 'lib/pgbus/streams/envelope.rb', line 58

def self.connected(id:)
  raise ArgumentError, "id is required" if id.nil? || id.to_s.empty?

  "event: pgbus:connected\ndata: #{JSON.generate({ connectionId: id.to_s })}\n\n"
end

.http_response_headers ⇒ Object 



72
73
74
 
# File 'lib/pgbus/streams/envelope.rb', line 72

def self.http_response_headers
  RESPONSE_HEADERS
end

.message(id:, event:, data:) ⇒ Object

Raises:

  • (ArgumentError) 


34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/pgbus/streams/envelope.rb', line 34

def self.message(id:, event:, data:)
  raise ArgumentError, "id is required" if id.nil?
  raise ArgumentError, "event is required" if event.nil? || event.to_s.empty?

  # The event name is a single SSE field line, so newlines are stripped —
  # an unescaped \r/\n would terminate the field early and let a crafted
  # value inject extra SSE fields (a forged id:/data:) into the frame.
  # The payload is framed as one `data:` line per payload line instead:
  # every line carries the `data: ` prefix, which is both spec-correct
  # (the client rejoins with \n) and injection-safe.
  "id: #{id}\nevent: #{strip_newlines(event.to_s)}\n#{data_lines(data.to_s)}\n"
end

.retry_directive(milliseconds) ⇒ Object 



64
65
66
67
68
69
70
 
# File 'lib/pgbus/streams/envelope.rb', line 64

def self.retry_directive(milliseconds)
  unless milliseconds.is_a?(Integer) && !milliseconds.negative?
    raise ArgumentError, "retry must be a non-negative integer (got #{milliseconds.inspect})"
  end

  "retry: #{milliseconds}\n\n"
end