Class: OMQ::QoS::DedupSet

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

Overview

Per-receiving-connection dedup set for QoS >= 2.

An ordered map of {digest => added_at} keyed by the 8-byte XXH64 / SHA-1 digest. Hash insertion order gives us oldest-first eviction for free.

Entries are added on message delivery, removed eagerly on CLR from the sender, and swept lazily on TTL expiry via #sweep. When a new entry would exceed capacity the oldest entry is evicted — the sender will not retransmit past dead_letter_timeout anyway, so an eviction older than that is safe.

Instance Method Summary collapse

Constructor Details

#initialize(capacity:) ⇒ DedupSet

Returns a new instance of DedupSet.

Parameters:

  • capacity (Integer) —

    typically recv_hwm; maximum entries before oldest-first eviction kicks in



22
23
24
25
 
# File 'lib/omq/qos/dedup_set.rb', line 22

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

Instance Method Details

#add(digest) ⇒ Object

Adds digest and stamps it with the current monotonic clock. Evicts the oldest entry if capacity would be exceeded.



36
37
38
39
40
41
 
# File 'lib/omq/qos/dedup_set.rb', line 36

def add(digest)
  if @entries.size >= @capacity && !@entries.key?(digest)
    @entries.shift
  end
  @entries[digest] = Async::Clock.now
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


64
65
66
 
# File 'lib/omq/qos/dedup_set.rb', line 64

def empty?
  @entries.empty?
end

#remove(digest) ⇒ Object

Removes an entry (called on CLR from the sender).



45
46
47
 
# File 'lib/omq/qos/dedup_set.rb', line 45

def remove(digest)
  @entries.delete(digest)
end

#seen?(digest) ⇒ Boolean

Parameters:

  • digest (String) —

    8-byte binary

Returns:

  • (Boolean) 


29
30
31
 
# File 'lib/omq/qos/dedup_set.rb', line 29

def seen?(digest)
  @entries.key?(digest)
end

#size ⇒ Integer

Returns:

  • (Integer) 


58
59
60
 
# File 'lib/omq/qos/dedup_set.rb', line 58

def size
  @entries.size
end

#sweep(now, ttl) ⇒ Object

Drops entries older than ttl seconds at now.



51
52
53
54
 
# File 'lib/omq/qos/dedup_set.rb', line 51

def sweep(now, ttl)
  cutoff = now - ttl
  @entries.delete_if { |_digest, added_at| added_at < cutoff }
end