Class: Pgbus::Web::Streamer::Connection

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

Overview

Wraps a single hijacked SSE client socket with its own cursor state, per-io mutex, and liveness flag. Owns no threads — the Dispatcher and Heartbeat threads call #enqueue / #write_comment on Connection instances directly, and the per-io mutex in IoWriter serialises concurrent writes.

Cursor semantics: last_msg_id_sent is strictly monotonic. enqueue filters envelopes with msg_id > last_msg_id_sent and advances the cursor only for envelopes that actually wrote successfully. This is the client-side leg of the replay-race fix (§6.5 of the design doc).

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(id:, stream_name:, io:, since_id:, writer:, write_deadline_ms:, context: nil) ⇒ Connection

Returns a new instance of Connection.



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/pgbus/web/streamer/connection.rb', line 34

def initialize(id:, stream_name:, io:, since_id:, writer:, write_deadline_ms:, context: nil)
  @id = id
  @stream_name = stream_name
  @io = io
  @last_msg_id_sent = Concurrent::AtomicReference.new(since_id.to_i)
  @writer = writer
  @write_deadline_ms = write_deadline_ms
  @mutex = Mutex.new
  @dead = false
  @closed = false
  @presence_member = nil
  @created_at = monotonic
  @last_write_at = @created_at
  # Context is whatever the StreamApp's authorize hook returned
  # (a truthy non-boolean value). Typically a user model or a
  # session hash. The Dispatcher passes it to the Filters
  # registry when evaluating visible_to predicates. Defaults to
  # nil for tests that don't need audience filtering.
  @context = context
end

Instance Attribute Details

#context ⇒ Object (readonly)

Returns the value of attribute context.



16
17
18
 
# File 'lib/pgbus/web/streamer/connection.rb', line 16

def context
  @context
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



16
17
18
 
# File 'lib/pgbus/web/streamer/connection.rb', line 16

def id
  @id
end

#io ⇒ Object (readonly)

Returns the value of attribute io.



16
17
18
 
# File 'lib/pgbus/web/streamer/connection.rb', line 16

def io
  @io
end

#mutex ⇒ Object (readonly)

Returns the value of attribute mutex.



16
17
18
 
# File 'lib/pgbus/web/streamer/connection.rb', line 16

def mutex
  @mutex
end

#presence_member ⇒ Object

The presence member id this connection auto-joined as, or nil for non-presence streams / anonymous connections. Set by the Dispatcher on connect; read on disconnect and heartbeat touch.



20
21
22
 
# File 'lib/pgbus/web/streamer/connection.rb', line 20

def presence_member
  @presence_member
end

#stream_name ⇒ Object (readonly)

Returns the value of attribute stream_name.



16
17
18
 
# File 'lib/pgbus/web/streamer/connection.rb', line 16

def stream_name
  @stream_name
end

Instance Method Details

#close ⇒ Object

Idempotent socket close for use by Instance#shutdown! and the heartbeat idle reaper. Wraps the respond_to? / closed? dance so callers don't need to know about StringIO-in-tests vs real Socket-in-prod or about the mark_dead! ordering.

Takes the same mutex as IoWriter.write so it can't fire mid-write — otherwise the write loop could hit a half-closed socket and corrupt the last_msg_id_sent cursor by marking the connection dead between successful writes. The rescue narrows to IO-related exceptions; unrelated errors (bugs in the fake IO used by tests, nil-dereferences, etc.) should still propagate so the test suite catches them.



123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/pgbus/web/streamer/connection.rb', line 123

def close
  @mutex.synchronize do
    return if @closed

    @closed = true
    mark_dead!
    return unless @io.respond_to?(:close)

    @io.close unless @io.respond_to?(:closed?) && @io.closed?
  end
rescue IOError, SystemCallError => e
  Pgbus.logger&.debug { "[Pgbus::Streamer::Connection] close failed: #{e.class}: #{e.message}" }
end

#dead? ⇒ Boolean

Returns:

  • (Boolean) 


103
104
105
 
# File 'lib/pgbus/web/streamer/connection.rb', line 103

def dead?
  @dead
end

#enqueue(envelopes, deadline_ms: @write_deadline_ms) ⇒ Object

deadline_ms defaults to the connection's own write deadline so every existing caller is unchanged. The Dispatcher overrides it with the short streams_fanout_write_deadline_ms for hot-loop fanout writes, bounding head-of-line blocking on a slow client (issue #315 item 3).



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/pgbus/web/streamer/connection.rb', line 59

def enqueue(envelopes, deadline_ms: @write_deadline_ms)
  written = []
  envelopes.each do |envelope|
    ephemeral = envelope.msg_id.negative?
    next if !ephemeral && envelope.msg_id <= @last_msg_id_sent.get

    bytes = Pgbus::Streams::Envelope.message(
      id: envelope.msg_id,
      event: sse_event_for(envelope),
      data: envelope.payload
    )

    result = @writer.write(self, bytes, deadline_ms: deadline_ms)
    if result == :ok
      @last_msg_id_sent.set(envelope.msg_id) unless ephemeral
      @last_write_at = monotonic
      written << envelope
    else
      mark_dead!
      break
    end
  end
  written
end

#idle_for ⇒ Object 



99
100
101
 
# File 'lib/pgbus/web/streamer/connection.rb', line 99

def idle_for
  monotonic - @last_write_at
end

#last_msg_id_sent ⇒ Object

last_msg_id_sent is the ONE field whose writer can cross threads: with streams_writer_threads > 0 (issue #321) the OutboundPump worker thread advances it inside #enqueue, while the dispatcher thread reads it via cursor_for / handle_connect. A Concurrent::AtomicReference gives a real happens-before on every Ruby engine (not just MRI's GVL) at ~ns cost — one worker only ever writes a given connection's field (stable partition), so there's no writer-writer contention. Inline mode (default) reads/writes it single-threaded, semantically identical.



30
31
32
 
# File 'lib/pgbus/web/streamer/connection.rb', line 30

def last_msg_id_sent
  @last_msg_id_sent.get
end

#mark_dead! ⇒ Object 



107
108
109
 
# File 'lib/pgbus/web/streamer/connection.rb', line 107

def mark_dead!
  @dead = true
end

#write_comment(text) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
 
# File 'lib/pgbus/web/streamer/connection.rb', line 84

def write_comment(text)
  bytes = Pgbus::Streams::Envelope.comment(text)
  result = @writer.write(self, bytes, deadline_ms: @write_deadline_ms)
  if result == :ok
    @last_write_at = monotonic
  else
    mark_dead!
  end
  result
end

#write_sentinel(bytes) ⇒ Object 



95
96
97
 
# File 'lib/pgbus/web/streamer/connection.rb', line 95

def write_sentinel(bytes)
  @writer.write(self, bytes, deadline_ms: @write_deadline_ms)
end