Class: OMQ::QoS::PeerRegistry

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/qos/peer_registry.rb

Overview

Tracks sent-but-unacknowledged messages per peer for QoS >= 2.

Keyed by Protocol::ZMTP::PeerInfo so entries survive reconnects: when the same peer comes back, #resume returns the peer's pending entries for retransmit on the new connection. When a peer stays away longer than dead_letter_timeout, #sweep_dead_letters drains its entries as DeadLetters.

Backpressure lives on the owning OMQ::QoS instance (an Async::Semaphore bounded at send_hwm). The registry itself is a pure peer-indexed store; it does not block.

Defined Under Namespace

Classes: Entry, PeerState

Instance Method Summary collapse

Constructor Details

#initialize(capacity:) ⇒ PeerRegistry

Returns a new instance of PeerRegistry.

Parameters:

  • capacity (Integer) —

    send_hwm (kept for symmetry with PendingStore; the actual bound is enforced by the OMQ::QoS#send_semaphore)



47
48
49
50
 
# File 'lib/omq/qos/peer_registry.rb', line 47

def initialize(capacity:)
  @capacity = capacity
  @peers    = {}
end

Instance Method Details

#ack(peer_info, digest) ⇒ Object

Removes and returns the entry matching digest for peer_info, or nil if unknown (e.g. late ACK after dead-letter).



63
64
65
 
# File 'lib/omq/qos/peer_registry.rb', line 63

def ack(peer_info, digest)
  @peers[peer_info]&.entries&.delete(digest)
end

#connection_for(peer_info) ⇒ Protocol::ZMTP::Connection?

Returns the live connection currently pinned to peer_info, if any.

Returns:

  • (Protocol::ZMTP::Connection, nil) —

    the live connection currently pinned to peer_info, if any



70
71
72
 
# File 'lib/omq/qos/peer_registry.rb', line 70

def connection_for(peer_info)
  @peers[peer_info]&.connection
end

#disconnect(peer_info) ⇒ Object

Marks the peer as disconnected. Entries stay in place until either the peer reconnects (#resume) or the dead-letter timer fires (#sweep_dead_letters).



78
79
80
81
82
 
# File 'lib/omq/qos/peer_registry.rb', line 78

def disconnect(peer_info)
  state = @peers[peer_info] or return
  state.connection      = nil
  state.disconnected_at = Async::Clock.now
end

#drain_with_dead_letter(reason) ⇒ Object

Resolves all outstanding promises with a DeadLetter carrying reason. Called from OMQ::QoS#shutdown so no +#wait+-ing fiber hangs.



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/omq/qos/peer_registry.rb', line 117

def drain_with_dead_letter(reason)
  @peers.each do |peer_info, state|
    state.entries.each_value do |entry|
      next if entry.promise.resolved?
      dl = DeadLetter.new(
        parts:     entry.parts,
        reason:    reason,
        peer_info: peer_info,
        error:     nil,
      )
      entry.promise.resolve(dl)
    end
  end
  @peers.clear
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


141
142
143
 
# File 'lib/omq/qos/peer_registry.rb', line 141

def empty?
  @peers.all? { |_peer, state| state.entries.empty? }
end

#resume(peer_info, conn) ⇒ Object

Rebinds conn to peer_info and returns the existing pending entries in insertion order. Caller retransmits them on conn.



87
88
89
90
91
92
 
# File 'lib/omq/qos/peer_registry.rb', line 87

def resume(peer_info, conn)
  state                 = (@peers[peer_info] ||= PeerState.new(conn))
  state.connection      = conn
  state.disconnected_at = nil
  state.entries.values
end

#size ⇒ Integer

Returns total pending entries across all peers.

Returns:

  • (Integer) —

    total pending entries across all peers



135
136
137
 
# File 'lib/omq/qos/peer_registry.rb', line 135

def size
  @peers.sum { |_peer, state| state.entries.size }
end

#sweep_dead_letters(now, timeout) ⇒ Array<Array(Entry, Protocol::ZMTP::PeerInfo)>

Returns all entries whose peer has been gone for at least timeout seconds, pairs them with their peer_info, and removes their PeerState from the registry.

Parameters:

  • now (Float)
  • timeout (Numeric)

Returns:

  • (Array<Array(Entry, Protocol::ZMTP::PeerInfo)>) 


102
103
104
105
106
107
108
109
110
111
 
# File 'lib/omq/qos/peer_registry.rb', line 102

def sweep_dead_letters(now, timeout)
  expired = []
  @peers.delete_if do |peer_info, state|
    next false unless state.disconnected_at
    next false if (now - state.disconnected_at) < timeout
    state.entries.each_value { |entry| expired << [entry, peer_info] }
    true
  end
  expired
end

#track(peer_info, digest, entry, conn) ⇒ Object

Records a fresh pending entry, pinned to peer_info on conn.



54
55
56
57
58
 
# File 'lib/omq/qos/peer_registry.rb', line 54

def track(peer_info, digest, entry, conn)
  state = (@peers[peer_info] ||= PeerState.new(conn))
  state.connection = conn
  state.entries[digest] = entry
end