Class: OMQ::QoS::PendingStore

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

Overview

Tracks sent-but-unacknowledged messages for QoS 1.

Keyed by 8-byte XXH3 digest of the raw ZMTP wire bytes.

Bounded by an internal Async::Semaphore of capacity permits (typically send_hwm). The sender acquires a permit via #wait_for_slot before each #track; #ack and #messages_for release permits back. This gives the QoS path the same backpressure semantics the send queue has at QoS 0: once send_hwm messages are outstanding, the next acquire blocks until an ACK (or a connection drop that drains the entries) frees a slot.

Defined Under Namespace

Classes: Entry

Instance Method Summary collapse

Constructor Details

#initialize(capacity:) ⇒ PendingStore

Returns a new instance of PendingStore.

Parameters:

  • capacity (Integer) —

    max pending entries



25
26
27
28
29
 
# File 'lib/omq/qos/pending_store.rb', line 25

def initialize(capacity:)
  @entries   = {}
  @capacity  = capacity
  @semaphore = Async::Semaphore.new(capacity)
end

Instance Method Details

#ack(hash) ⇒ Entry?

Acknowledges a message. Releases one semaphore permit when the entry existed.

Returns:



53
54
55
56
57
 
# File 'lib/omq/qos/pending_store.rb', line 53

def ack(hash)
  entry = @entries.delete(hash)
  @semaphore.release if entry
  entry
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


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

def empty?
  @entries.empty?
end

#messages_for(connection) ⇒ Array<Entry>

Returns and removes all pending entries for a connection. Releases one semaphore permit per removed entry.

Returns:



64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/omq/qos/pending_store.rb', line 64

def messages_for(connection)
  removed = []
  @entries.delete_if do |_hash, entry|
    if entry.connection.equal?(connection)
      removed << entry
      true
    end
  end
  removed.size.times { @semaphore.release }
  removed
end

#size ⇒ Integer

Returns:

  • (Integer) 


78
79
80
 
# File 'lib/omq/qos/pending_store.rb', line 78

def size
  @entries.size
end

#track(hash, parts, connection) ⇒ Object 



40
41
42
43
44
45
46
 
# File 'lib/omq/qos/pending_store.rb', line 40

def track(hash, parts, connection)
  @entries[hash] = Entry.new(
    parts:      parts,
    connection: connection,
    sent_at:    Async::Clock.now,
  )
end

#wait_for_slot ⇒ Object

Blocks the caller until a pending-slot permit is available and then acquires it. Caller MUST follow up with a #track; the corresponding #ack (or #messages_for drop) releases.



35
36
37
 
# File 'lib/omq/qos/pending_store.rb', line 35

def wait_for_slot
  @semaphore.acquire
end