Class: AceMQ::AMQP::Patterns::InMemoryIdempotencyStore

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/idempotency.rb

Overview

An idempotency store that remembers keys in this process.

Right behind one worker, and wrong the moment there are two: each has its own memory, so both are told they are first and the duplicate goes through. It is also lost on restart, which turns every message in flight into a duplicate.

Use it in tests, and in a single-process service. Anything else wants a store the workers share — the same database the work is written to, in the same transaction, which is the only arrangement that actually holds.

Constant Summary collapse

DEFAULT_WINDOW =

How long a key is remembered when no window is given.

3600.0

Instance Method Summary collapse

Constructor Details

#initialize(window: DEFAULT_WINDOW) ⇒ InMemoryIdempotencyStore

The window is also the claim's lease: a key claimed and never confirmed is in progress until it ages out, and then it is free again.

Parameters:

  • window (Numeric) (defaults to: DEFAULT_WINDOW) —

    seconds to remember a key for. There has to be one: without it the map grows for as long as the process lives. Make it comfortably longer than the longest a message can take to stop being retried, because a key forgotten too early is a duplicate that gets through.



100
101
102
103
104
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 100

def initialize(window: DEFAULT_WINDOW)
  @window = window.positive? ? window.to_f : DEFAULT_WINDOW
  @seen = {}
  @lock = Mutex.new
end

Instance Method Details

#claim(key) ⇒ Symbol

Claims a key, or says why not: done already, or still being done.

Parameters:

  • key (String)

Returns:

  • (Symbol) —

    one of the Claim values



110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 110

def claim(key)
  key = key.to_s
  @lock.synchronize do
    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    sweep(now)
    held = @seen[key]
    next held[1] ? Claim::DUPLICATE : Claim::IN_PROGRESS if held

    @seen[key] = [now, false]
    Claim::CLAIMED
  end
end

#confirm(key) ⇒ Object

Records that the claimed message really was handled, so a repeat is a duplicate rather than work in progress.



131
132
133
134
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 131

def confirm(key)
  @lock.synchronize { @seen[key.to_s]&.[]=(1, true) }
  nil
end

#first_time?(key) ⇒ Boolean

Records a key, and says whether it is new.

Parameters:

  • key (String)

Returns:

  • (Boolean) —

    true the first time, false for every repeat



127
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 127

def first_time?(key) = claim(key) == Claim::CLAIMED

#forget(key) ⇒ Object

Removes a key, so the message it belongs to can be handled again.



137
138
139
140
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 137

def forget(key)
  @lock.synchronize { @seen.delete(key.to_s) }
  nil
end

#size ⇒ Object

How many keys are remembered, for a test that wants to know the window is doing its job.



144
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 144

def size = @lock.synchronize { @seen.size }