Class: Pgbus::Web::Streamer::FalconConnection

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

Overview

Connection adapter for Falcon's native streaming body path. Wraps a Protocol::HTTP::Body::Writable instead of a raw IO socket.

Satisfies the same duck-type interface as Connection so the Dispatcher, Heartbeat, and Instance can use either type interchangeably. Key difference: no IoWriter — writes go directly to body.write which is backed by Thread::Queue and is fiber-safe under Falcon's scheduler.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(id:, stream_name:, body:, since_id:, write_deadline_ms:, context: nil) ⇒ FalconConnection

Returns a new instance of FalconConnection.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 26

def initialize(id:, stream_name:, body:, since_id:, write_deadline_ms:, context: nil)
  @id = id
  @stream_name = stream_name
  @body = body
  @io = body
  @last_msg_id_sent = Concurrent::AtomicReference.new(since_id.to_i)
  @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 = context
end

Instance Attribute Details

#context ⇒ Object (readonly)

Returns the value of attribute context.



15
16
17
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 15

def context
  @context
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



15
16
17
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 15

def id
  @id
end

#io ⇒ Object (readonly)

Returns the value of attribute io.



15
16
17
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 15

def io
  @io
end

#mutex ⇒ Object (readonly)

Returns the value of attribute mutex.



15
16
17
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 15

def mutex
  @mutex
end

#presence_member ⇒ Object

Returns the value of attribute presence_member.



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

def presence_member
  @presence_member
end

#stream_name ⇒ Object (readonly)

Returns the value of attribute stream_name.



15
16
17
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 15

def stream_name
  @stream_name
end

Instance Method Details

#close ⇒ Object 



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

def close
  @mutex.synchronize do
    return if @closed

    @closed = true
    mark_dead!
    @body.close_write
  end
rescue StandardError => e
  Pgbus.logger&.debug { "[Pgbus::Streamer::FalconConnection] close failed: #{e.class}: #{e.message}" }
end

#dead? ⇒ Boolean

Returns:

  • (Boolean) 


101
102
103
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 101

def dead?
  @dead || @body.closed?
end

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

deadline_ms is accepted for duck-type parity with Connection#enqueue (so the Dispatcher's safe_enqueue can call either class) but ignored: write_to_body is non-blocking under Falcon's fiber reactor, so there is no head-of-line blocking to bound here (issue #315 item 3).



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 46

def enqueue(envelopes, deadline_ms: @write_deadline_ms) # rubocop:disable Lint/UnusedMethodArgument
  written = []
  envelopes.each do |envelope|
    next if 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 = write_to_body(bytes)
    if result == :ok
      @last_msg_id_sent.set(envelope.msg_id)
      @last_write_at = monotonic
      written << envelope
    else
      mark_dead!
      break
    end
  end
  written
end

#idle_for ⇒ Object 



97
98
99
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 97

def idle_for
  monotonic - @last_write_at
end

#last_msg_id_sent ⇒ Object

last_msg_id_sent mirrors Connection's cross-thread cursor (issue #321): a Concurrent::AtomicReference so the OutboundPump worker's advance and the dispatcher's read have a happens-before on every Ruby engine. See the note on Connection#last_msg_id_sent.



22
23
24
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 22

def last_msg_id_sent
  @last_msg_id_sent.get
end

#mark_dead! ⇒ Object 



105
106
107
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 105

def mark_dead!
  @dead = true
end

#write_comment(text) ⇒ Object 



70
71
72
73
74
75
76
77
78
79
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 70

def write_comment(text)
  bytes = Pgbus::Streams::Envelope.comment(text)
  result = write_to_body(bytes)
  if result == :ok
    @last_write_at = monotonic
  else
    mark_dead!
  end
  result
end

#write_sentinel(bytes) ⇒ Object 



81
82
83
 
# File 'lib/pgbus/web/streamer/falcon_connection.rb', line 81

def write_sentinel(bytes)
  write_to_body(bytes)
end