Class: OMQ::QoS

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/qos.rb,
lib/omq/qos/hasher.rb,
lib/omq/qos/version.rb,
lib/omq/qos/dedup_set.rb,
lib/omq/qos/socket_ext.rb,
lib/omq/qos/dead_letter.rb,
lib/omq/qos/error_codes.rb,
lib/omq/qos/options_ext.rb,
lib/omq/qos/routing_ext.rb,
lib/omq/qos/lifecycle_ext.rb,
lib/omq/qos/peer_registry.rb,
lib/omq/qos/pending_store.rb,
lib/omq/qos/connection_ext.rb,
lib/omq/qos/retry_scheduler.rb,
lib/omq/qos/zmtp/command_ext.rb

Overview

Per-hop delivery guarantees (QoS) extension for OMQ sockets.

Defined Under Namespace

Modules: CommandClassExt, CommandExt, ConnectionExt, ErrorCodes, GatherExt, LifecycleExt, OptionsExt, PullExt, PushExt, ReadableExt, RetryScheduler, RoundRobinExt, ScatterExt, SocketExt, WritableExt Classes: BadInputError, DeadLetter, DedupSet, Envelope, NackError, NackInfo, OverloadedError, PeerRegistry, PendingStore, RejectedError, TimeoutError

Constant Summary collapse

DEFAULT_DEAD_LETTER_TIMEOUT = 
60.0
DEFAULT_DEDUP_TTL = 
60.0
DEFAULT_MAX_RETRIES = 
3
DEFAULT_RETRY_BACKOFF = 
(0.1..10.0).freeze
SUPPORTED_HASH_ALGOS =

Supported hash algorithms in default preference order. x = XXH64 (8 bytes) — fast, requires xxhash library s = SHA-1 truncated to 64 bits (8 bytes) — available in any standard library

All digests are 8 bytes (64 bits). Collision probability within the in-flight window (typically ≤ HWM) is negligible: ~N²/2⁶⁵ where N is the number of un-ACK'd messages.

"xs".freeze
DEFAULT_HASH_ALGO = 
"x".freeze
HASH_SIZE = 
8
VERSION = 
"0.3.2"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(level:, hash_algos: SUPPORTED_HASH_ALGOS, dead_letter_timeout: nil, dedup_ttl: nil, max_retries: nil, processing_timeout: nil, retry_backoff: nil) ⇒ QoS

Constructed only via the class-method builders.



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/omq/qos.rb', line 146

def initialize(level:,
               hash_algos:          SUPPORTED_HASH_ALGOS,
               dead_letter_timeout: nil,
               dedup_ttl:           nil,
               max_retries:         nil,
               processing_timeout:  nil,
               retry_backoff:       nil)
  @level               = level
  @hash_algos          = hash_algos
  @dead_letter_timeout = dead_letter_timeout
  @dedup_ttl           = dedup_ttl
  @max_retries         = max_retries
  @processing_timeout  = processing_timeout
  @retry_backoff       = retry_backoff

  @engine            = nil
  @peer_registry     = nil
  @send_semaphore    = nil
  @dedup_sets        = nil
  @sweep_task        = nil
  @enqueue_promises  = nil
end

Instance Attribute Details

#dead_letter_timeout ⇒ Object (readonly)

Returns the value of attribute dead_letter_timeout.



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

def dead_letter_timeout
  @dead_letter_timeout
end

#dedup_ttl ⇒ Object (readonly)

Returns the value of attribute dedup_ttl.



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

def dedup_ttl
  @dedup_ttl
end

#hash_algos ⇒ Object (readonly)

Returns the value of attribute hash_algos.



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

def hash_algos
  @hash_algos
end

#level ⇒ Object (readonly)

Returns the value of attribute level.



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

def level
  @level
end

#max_retries ⇒ Object (readonly)

Returns the value of attribute max_retries.



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

def max_retries
  @max_retries
end

#processing_timeout ⇒ Object (readonly)

Returns the value of attribute processing_timeout.



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

def processing_timeout
  @processing_timeout
end

#retry_backoff ⇒ Object (readonly)

Returns the value of attribute retry_backoff.



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

def retry_backoff
  @retry_backoff
end

Class Method Details

.ack_command(parts, algorithm: DEFAULT_HASH_ALGO) ⇒ Protocol::ZMTP::Codec::Command

Builds an ACK command for the given message.

Parameters:

  • parts (Array<String>) —

    message frames

  • algorithm (String) (defaults to: DEFAULT_HASH_ALGO) —

    "x" (XXH64) or "s" (SHA-1 truncated)

Returns:

  • (Protocol::ZMTP::Codec::Command) 


60
61
62
 
# File 'lib/omq/qos/hasher.rb', line 60

def self.ack_command(parts, algorithm: DEFAULT_HASH_ALGO)
  Protocol::ZMTP::Codec::Command.ack(digest(parts, algorithm: algorithm), algorithm: algorithm)
end

.algo_for(connection) ⇒ String

Picks a hash algorithm for a connection by intersecting our preferences with the peer's advertised list (read from the peer's READY properties).

Parameters:

  • connection (Protocol::ZMTP::Connection)

Returns:

  • (String) —

    single-char algorithm identifier



164
165
166
167
 
# File 'lib/omq/qos/routing_ext.rb', line 164

def self.algo_for(connection)
  peer = connection.respond_to?(:peer_properties) ? connection.peer_properties : nil
  negotiate_hash(peer&.fetch("X-QoS-Hash", "") || "") || DEFAULT_HASH_ALGO
end

.at_least_once(hash_algos: SUPPORTED_HASH_ALGOS) ⇒ QoS

Builds a QoS 1 (at-least-once) handle.

Parameters:

  • hash_algos (String) (defaults to: SUPPORTED_HASH_ALGOS) —

    preference list, e.g. "xs"

Returns:



84
85
86
 
# File 'lib/omq/qos.rb', line 84

def self.at_least_once(hash_algos: SUPPORTED_HASH_ALGOS)
  new(level: 1, hash_algos: hash_algos)
end

.dead_letter(entry, peer_info:, reason:, semaphore: nil, error: nil) ⇒ DeadLetter?

Builds a DeadLetter for entry, resolves the entry's Promise with it, and releases one slot on semaphore so a waiting sender can proceed.

No-op if the Promise is already resolved (late ACK after dead-letter, or shutdown racing a sweep).

Parameters:

  • entry (PeerRegistry::Entry)
  • peer_info (Protocol::ZMTP::PeerInfo)
  • reason (Symbol) —

    :peer_timeout, :terminal_nack, :retry_exhausted, or :socket_closed

  • semaphore (Async::Semaphore, nil) (defaults to: nil) —

    the send-slot semaphore to release; pass nil on shutdown-drain where the whole semaphore is being torn down anyway

  • error (Object, nil) (defaults to: nil) —

    NackInfo at QoS 3; nil at QoS 2

Returns:

  • (DeadLetter, nil) —

    the built DeadLetter, or nil if the Promise was already resolved



22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/omq/qos/dead_letter.rb', line 22

def self.dead_letter(entry, peer_info:, reason:, semaphore: nil, error: nil)
  return nil if entry.promise.resolved?

  dl = DeadLetter.new(
    parts:     entry.parts,
    reason:    reason,
    peer_info: peer_info,
    error:     error,
  )
  entry.promise.resolve(dl)
  semaphore&.release
  dl
end

.digest(parts, algorithm: DEFAULT_HASH_ALGO) ⇒ String

Computes an 8-byte digest over raw ZMTP wire bytes.

Parameters:

  • parts (Array<String>) —

    message frames

  • algorithm (String) (defaults to: DEFAULT_HASH_ALGO) —

    "x" (XXH64) or "s" (SHA-1 truncated)

Returns:

  • (String) —

    8-byte binary digest



26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/omq/qos/hasher.rb', line 26

def self.digest(parts, algorithm: DEFAULT_HASH_ALGO)
  wire = Protocol::ZMTP::Codec::Frame.encode_message(parts)
  case algorithm
  when "x"
    [XXhash.xxh64(wire)].pack("Q<")
  when "s"
    Digest::SHA1.digest(wire).byteslice(0, 8)
  else
    raise ArgumentError, "unsupported QoS hash algorithm: #{algorithm.inspect}"
  end
end

.exactly_once(hash_algos: SUPPORTED_HASH_ALGOS, dead_letter_timeout: DEFAULT_DEAD_LETTER_TIMEOUT, dedup_ttl: DEFAULT_DEDUP_TTL) ⇒ QoS

Builds a QoS 2 (exactly-once with peer pinning) handle.

Parameters:

  • hash_algos (String) (defaults to: SUPPORTED_HASH_ALGOS)
  • dead_letter_timeout (Numeric) (defaults to: DEFAULT_DEAD_LETTER_TIMEOUT) —

    seconds to wait for a disconnected peer to return before dead-lettering its pending messages

  • dedup_ttl (Numeric) (defaults to: DEFAULT_DEDUP_TTL) —

    seconds a receiver keeps a digest in its dedup set before it may evict it

Returns:



97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/omq/qos.rb', line 97

def self.exactly_once(
  hash_algos:           SUPPORTED_HASH_ALGOS,
  dead_letter_timeout:  DEFAULT_DEAD_LETTER_TIMEOUT,
  dedup_ttl:            DEFAULT_DEDUP_TTL
)
  new(
    level:               2,
    hash_algos:          hash_algos,
    dead_letter_timeout: dead_letter_timeout,
    dedup_ttl:           dedup_ttl,
  )
end

.exactly_once_and_processed(hash_algos: SUPPORTED_HASH_ALGOS, dead_letter_timeout: DEFAULT_DEAD_LETTER_TIMEOUT, dedup_ttl: DEFAULT_DEDUP_TTL, max_retries: DEFAULT_MAX_RETRIES, processing_timeout: nil, retry_backoff: DEFAULT_RETRY_BACKOFF) ⇒ QoS

Builds a QoS 3 (exactly-once + application-level COMP/NACK) handle.

Parameters:

  • hash_algos (String) (defaults to: SUPPORTED_HASH_ALGOS)
  • dead_letter_timeout (Numeric) (defaults to: DEFAULT_DEAD_LETTER_TIMEOUT)
  • dedup_ttl (Numeric) (defaults to: DEFAULT_DEDUP_TTL)
  • max_retries (Integer) (defaults to: DEFAULT_MAX_RETRIES) —

    retry cap on retryable NACKs

  • processing_timeout (Numeric, nil) (defaults to: nil) —

    per-message receiver handler deadline

  • retry_backoff (Range) (defaults to: DEFAULT_RETRY_BACKOFF) —

    exponential backoff bounds (min..max, seconds)

Returns:



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/omq/qos.rb', line 120

def self.exactly_once_and_processed(
  hash_algos:           SUPPORTED_HASH_ALGOS,
  dead_letter_timeout:  DEFAULT_DEAD_LETTER_TIMEOUT,
  dedup_ttl:            DEFAULT_DEDUP_TTL,
  max_retries:          DEFAULT_MAX_RETRIES,
  processing_timeout:   nil,
  retry_backoff:        DEFAULT_RETRY_BACKOFF
)
  new(
    level:               3,
    hash_algos:          hash_algos,
    dead_letter_timeout: dead_letter_timeout,
    dedup_ttl:           dedup_ttl,
    max_retries:         max_retries,
    processing_timeout:  processing_timeout,
    retry_backoff:       retry_backoff,
  )
end

.handle_qos3_nack(conn, qos, algo, peer_info, entry, hash, code, message) ⇒ void

This method returns an undefined value.

NACK dispatch for QoS 3 senders. Terminal codes dead-letter immediately; retryable codes either retry on the same peer after exponential backoff or dead-letter once max_retries is reached.



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/omq/qos/routing_ext.rb', line 89

def self.handle_qos3_nack(conn, qos, algo, peer_info, entry, hash, code, message)
  nack_info = NackInfo.new(code: code, message: message)

  unless ErrorCodes.retryable?(code)
    QoS.dead_letter(entry,
      peer_info: peer_info,
      reason:    :terminal_nack,
      semaphore: qos.send_semaphore,
      error:     nack_info,
    )
    return
  end

  if entry.retry_count + 1 >= qos.max_retries
    QoS.dead_letter(entry,
      peer_info: peer_info,
      reason:    :retry_exhausted,
      semaphore: qos.send_semaphore,
      error:     nack_info,
    )
    return
  end

  next_entry = entry.with(retry_count: entry.retry_count + 1, sent_at: Async::Clock.now)
  qos.peer_registry.track(peer_info, hash, next_entry, conn)

  delay = RetryScheduler.delay(entry.retry_count, qos.retry_backoff)
  qos.spawn_task(annotation: "qos3 retry") do
    sleep delay
    live_conn = qos.peer_registry.connection_for(peer_info)
    live_conn&.send_message(next_entry.parts)
  end
end

.handshake_metadata(options) ⇒ Hash{String => String}

Builds the READY-property metadata hash this socket should advertise based on its QoS configuration. Empty hash at QoS 0 so the Connection sees no extras.

Parameters:

  • options (OMQ::Options)

Returns:

  • (Hash{String => String}) 


55
56
57
58
59
60
61
 
# File 'lib/omq/qos/lifecycle_ext.rb', line 55

def self.handshake_metadata(options)
  qos = options.qos
  return {} if qos.nil?
  meta = { "X-QoS" => qos.level.to_s }
  meta["X-QoS-Hash"] = qos.hash_algos unless qos.hash_algos.empty?
  meta
end

.install_qos1_handler(conn, pending_store) ⇒ Object

Installs the QoS 1 command handler: dispatches incoming ACK frames to pending_store.ack.

Parameters:

  • conn (Protocol::ZMTP::Connection)
  • pending_store (PendingStore) 


12
13
14
15
16
17
18
 
# File 'lib/omq/qos/routing_ext.rb', line 12

def self.install_qos1_handler(conn, pending_store)
  conn.qos_on_command = lambda do |cmd|
    next unless cmd.name == "ACK"
    _, hash = cmd.ack_data
    pending_store.ack(hash)
  end
end

.install_qos2_receiver_handler(conn, dedup) ⇒ Object

Installs the QoS 2 receiver-side command handler on conn. Dispatches CLR → evict digest from the peer's dedup set.

Parameters:

  • conn (Protocol::ZMTP::Connection)
  • dedup (DedupSet) 


129
130
131
132
133
134
135
 
# File 'lib/omq/qos/routing_ext.rb', line 129

def self.install_qos2_receiver_handler(conn, dedup)
  conn.qos_on_command = lambda do |cmd|
    next unless cmd.name == "CLR"
    _, hash = cmd.clr_data
    dedup.remove(hash)
  end
end

.install_qos2_sender_handler(conn, qos, algo) ⇒ Object

Installs the QoS 2 sender-side command handler on conn. Dispatches ACK → resolve promise + send CLR + release a #send_semaphore permit.

Parameters:

  • conn (Protocol::ZMTP::Connection)
  • qos (OMQ::QoS)
  • algo (String) 


28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/omq/qos/routing_ext.rb', line 28

def self.install_qos2_sender_handler(conn, qos, algo)
  peer_info = conn.peer_info
  semaphore = qos.send_semaphore
  registry  = qos.peer_registry
  clr_class = Protocol::ZMTP::Codec::Command

  conn.qos_on_command = lambda do |cmd|
    next unless cmd.name == "ACK"
    _, hash = cmd.ack_data
    entry = registry.ack(peer_info, hash)
    next unless entry

    entry.promise.resolve(:delivered) unless entry.promise.resolved?
    semaphore.release

    conn.send_command(clr_class.clr(hash, algorithm: algo))
  end
end

.install_qos3_sender_handler(conn, qos, algo) ⇒ Object

Installs the QoS 3 sender-side command handler on conn. Dispatches COMP → resolve promise + send CLR + release a permit; NACK → classify code, either dead-letter terminally or schedule a retry with exponential backoff (up to max_retries).

Parameters:

  • conn (Protocol::ZMTP::Connection)
  • qos (OMQ::QoS)
  • algo (String) 


56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/omq/qos/routing_ext.rb', line 56

def self.install_qos3_sender_handler(conn, qos, algo)
  peer_info = conn.peer_info
  semaphore = qos.send_semaphore
  registry  = qos.peer_registry
  cmd_class = Protocol::ZMTP::Codec::Command

  conn.qos_on_command = lambda do |cmd|
    case cmd.name
    when "COMP"
      _, hash = cmd.comp_data
      entry = registry.ack(peer_info, hash)
      next unless entry

      entry.promise.resolve(:delivered) unless entry.promise.resolved?
      semaphore.release
      conn.send_command(cmd_class.clr(hash, algorithm: algo))

    when "NACK"
      _, hash, code, message = cmd.nack_data
      entry = registry.ack(peer_info, hash)
      next unless entry

      handle_qos3_nack(conn, qos, algo, peer_info, entry, hash, code, message)
    end
  end
end

.negotiate_hash(peer_hash) ⇒ String?

Negotiates the hash algorithm for a connection. Returns the first algo in our preference list that the peer supports.

Parameters:

  • peer_hash (String) —

    peer's supported algos (e.g. "sx")

Returns:

  • (String, nil) —

    single-char algorithm, or nil if no overlap



45
46
47
48
49
50
51
 
# File 'lib/omq/qos/hasher.rb', line 45

def self.negotiate_hash(peer_hash)
  return DEFAULT_HASH_ALGO if peer_hash.empty?
  SUPPORTED_HASH_ALGOS.each_char do |algo|
    return algo if peer_hash.include?(algo)
  end
  nil
end

.on_send_connection_added(socket, conn) ⇒ void

This method returns an undefined value.

Shared helper for the send-side connection_added path: installs the QoS 1 or QoS 2 command handler, and at QoS 2 replays any pending entries for the peer onto the fresh connection.

Parameters:

  • socket (OMQ::Routing::RoundRobin) —

    the routing strategy instance

  • conn (Protocol::ZMTP::Connection) 


283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
 
# File 'lib/omq/qos/routing_ext.rb', line 283

def self.on_send_connection_added(socket, conn)
  qos = socket.instance_variable_get(:@engine).options.qos
  return if qos.nil?
  return if reliable_transport?(conn)

  algo = socket.send(:algo_for, conn)

  case qos.level
  when 1
    install_qos1_handler(conn, socket.send(:pending_store))
  when 2
    install_qos2_sender_handler(conn, qos, algo)
    qos.ensure_sweep_task_started
    replay_pending(conn, qos)
  when 3
    install_qos3_sender_handler(conn, qos, algo)
    qos.ensure_sweep_task_started
    replay_pending(conn, qos)
  end
end

.process_qos3_envelope(env, qos) {|parts| ... } ⇒ Array<String>

Runs the application block against a QoS 3 Envelope and emits COMP on success, NACK on any StandardError.

Success path: add the digest to the peer's dedup set first, then send COMP — a late retransmit from the sender (crossing our COMP on the wire) must see a seen-digest so the receiver answers with a duplicate COMP rather than invoking the handler again.

Parameters:

Yield Parameters:

  • parts (Array<String>)

Returns:

  • (Array<String>) —

    the original message parts



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/omq/qos/socket_ext.rb', line 100

def self.process_qos3_envelope(env, qos, &block)
  parts  = env.parts
  conn   = env.conn
  digest = env.digest
  algo   = env.algo
  dedup  = qos.dedup_set_for(conn)

  if (timeout = qos.processing_timeout)
    Async::Task.current.with_timeout(timeout, TimeoutError) do
      block.call(parts)
    end
  else
    block.call(parts)
  end

  dedup.add(digest)
  conn.send_command(Protocol::ZMTP::Codec::Command.comp(digest, algorithm: algo))
  parts
rescue StandardError => error
  code, msg = ErrorCodes.exception_to_payload(error)
  conn.send_command(Protocol::ZMTP::Codec::Command.nack(digest, code: code, message: msg, algorithm: algo))
  parts
end

.reliable_transport?(conn) ⇒ Boolean

Inproc Pipes deliver messages synchronously through a shared in-memory queue, so there's nothing for at-least-once to protect against. QoS hooks short-circuit when they see a Pipe.

Parameters:

  • conn (Object)

Returns:

  • (Boolean) 


151
152
153
154
 
# File 'lib/omq/qos/routing_ext.rb', line 151

def self.reliable_transport?(conn)
  defined?(OMQ::Transport::Inproc::Pipe) &&
    conn.is_a?(OMQ::Transport::Inproc::Pipe)
end

.replay_pending(conn, qos) ⇒ Object

Replays any pending entries for conn.peer_info on conn, preserving original insertion order. Called on reconnect at QoS >= 2.

Parameters:

  • conn (Protocol::ZMTP::Connection)
  • qos (OMQ::QoS) 


311
312
313
314
315
316
317
318
319
 
# File 'lib/omq/qos/routing_ext.rb', line 311

def self.replay_pending(conn, qos)
  peer_info = conn.peer_info
  entries   = qos.peer_registry.resume(peer_info, conn)
  return if entries.empty?

  entries.each do |entry|
    conn.send_message(entry.parts)
  end
end

.validate_handshake!(options, conn) ⇒ Object

Validates the peer's READY properties against the local QoS configuration. At QoS >= 2 also asserts that the connection has a stable per-peer identity (CURVE pubkey or non-empty ZMQ_IDENTITY) — without one the PeerRegistry cannot pin pending messages across reconnects.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/omq/qos/lifecycle_ext.rb', line 69

def self.validate_handshake!(options, conn)
  props     = conn.peer_properties || {}
  qos       = options.qos
  local_lvl = qos&.level || 0
  peer_lvl  = (props["X-QoS"] || "0").to_i

  if local_lvl != peer_lvl
    raise Protocol::ZMTP::Error,
          "QoS mismatch: local=#{local_lvl} peer=#{peer_lvl}"
  end

  return if local_lvl == 0

  local_hash = qos.hash_algos
  peer_hash  = props["X-QoS-Hash"] || ""

  unless local_hash.empty? || peer_hash.empty?
    algo = local_hash.each_char.find { |c| peer_hash.include?(c) }
    unless algo
      raise Protocol::ZMTP::Error,
            "QoS hash algorithm mismatch: local=#{local_hash.inspect} peer=#{peer_hash.inspect}"
    end
  end

  return if local_lvl < 2

  info = conn.peer_info
  if info.nil? || (info.public_key.nil? && info.identity.empty?)
    raise Protocol::ZMTP::Error,
          "QoS #{local_lvl} requires a stable peer anchor " \
          "(CURVE public key or non-empty ZMQ_IDENTITY)"
  end
end

Instance Method Details

#attach!(engine) ⇒ void

This method returns an undefined value.

Binds this QoS handle to a socket's engine. Called by OMQ::QoS::SocketExt#qos= once the instance is assigned.

Parameters:

  • engine (OMQ::Engine)

Raises:

  • (ArgumentError) —

    if already attached to a (possibly different) engine



200
201
202
203
204
205
 
# File 'lib/omq/qos.rb', line 200

def attach!(engine)
  if @engine && !@engine.equal?(engine)
    raise ArgumentError, "OMQ::QoS instance is already attached to a different socket"
  end
  @engine = engine
end

#attached? ⇒ Boolean

Returns:

  • (Boolean) 


209
210
211
 
# File 'lib/omq/qos.rb', line 209

def attached?
  !@engine.nil?
end

#dedup_set_for(conn) ⇒ DedupSet

Lazily built per-connection dedup set for the receive side (QoS >= 2).

Parameters:

  • conn (Protocol::ZMTP::Connection)

Returns:



237
238
239
 
# File 'lib/omq/qos.rb', line 237

def dedup_set_for(conn)
  (@dedup_sets ||= {})[conn] ||= DedupSet.new(capacity: @engine.options.recv_hwm)
end

#enqueue_promise(parts, promise) ⇒ Object

Registers the Async::Promise associated with an outgoing parts array keyed by parts.object_id. The routing-layer OMQ::QoS::RoundRobinExt#write_batch looks it up (via #take_enqueue_promise) right before handing the entry to the PeerRegistry. Using object_id on a freshly-frozen Array that lives in the send queue avoids reshaping the send queue itself (which would otherwise need a Data wrapper every consumer has to know about).

Parameters:

  • parts (Array<String>)
  • promise (Async::Promise) 


181
182
183
 
# File 'lib/omq/qos.rb', line 181

def enqueue_promise(parts, promise)
  (@enqueue_promises ||= {})[parts.object_id] = promise
end

#ensure_sweep_task_started ⇒ void

This method returns an undefined value.

Starts the per-socket dead-letter sweep fiber (idempotent). Called from the routing layer once the first connection reaches the handshake-ready point — that's when Async::Task.current is guaranteed to be inside the engine's task tree.



277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
 
# File 'lib/omq/qos.rb', line 277

def ensure_sweep_task_started
  return if @sweep_task
  return unless @level >= 2 && @dead_letter_timeout

  interval = @dead_letter_timeout / 4.0
  @sweep_task = @engine.spawn_pump_task(annotation: "qos dead-letter sweep") do
    loop do
      sleep interval
      next unless @peer_registry

      expired = @peer_registry.sweep_dead_letters(Async::Clock.now, @dead_letter_timeout)
      expired.each do |entry, peer_info|
        QoS.dead_letter(entry, peer_info: peer_info, reason: :peer_timeout, semaphore: @send_semaphore)
      end
    end
  end
end

#forget_dedup_set(conn) ⇒ Object

Drops the dedup set for a connection (called on disconnect).



243
244
245
 
# File 'lib/omq/qos.rb', line 243

def forget_dedup_set(conn)
  @dedup_sets&.delete(conn)
end

#peer_registry ⇒ PeerRegistry

Lazily built pending-message registry (QoS >= 2). Keyed by Protocol::ZMTP::PeerInfo so entries survive reconnects.

Returns:



218
219
220
 
# File 'lib/omq/qos.rb', line 218

def peer_registry
  @peer_registry ||= PeerRegistry.new(capacity: @engine.options.send_hwm)
end

#send_semaphore ⇒ Async::Semaphore

Lazily built bounded-slot semaphore. Bounds the number of in-flight (un-ACK'd / un-COMP'd) messages at send_hwm; senders acquire a slot before tracking and release on ACK/COMP/dead-letter.

Returns:

  • (Async::Semaphore) 


228
229
230
 
# File 'lib/omq/qos.rb', line 228

def send_semaphore
  @send_semaphore ||= Async::Semaphore.new(@engine.options.send_hwm)
end

#shutdown ⇒ void

This method returns an undefined value.

Tears down socket-scoped state. Called from Socket#close. Resolves any pending Promises with DeadLetter(reason: :socket_closed) so no fiber hangs on #wait.



253
254
255
256
257
 
# File 'lib/omq/qos.rb', line 253

def shutdown
  @sweep_task&.stop
  @sweep_task = nil
  @peer_registry&.drain_with_dead_letter(:socket_closed)
end

#spawn_task(annotation:) { ... } ⇒ Async::Task

Spawns a task on the engine's task tree. Used by the QoS 3 retry scheduler and internal sweep fiber.

Parameters:

  • annotation (String)

Yields:

  • the task body

Returns:

  • (Async::Task) 


266
267
268
 
# File 'lib/omq/qos.rb', line 266

def spawn_task(annotation:, &block)
  @engine.spawn_pump_task(annotation: annotation, &block)
end

#take_enqueue_promise(parts) ⇒ Object

Removes and returns the Promise previously registered for parts. nil if none (e.g. message was enqueued before QoS upgrade — should not happen under the rules enforced by OMQ::QoS::SocketExt#qos=).



189
190
191
 
# File 'lib/omq/qos.rb', line 189

def take_enqueue_promise(parts)
  @enqueue_promises&.delete(parts.object_id)
end