Class: Engram::Adapters::InMemoryProcessedTurns

Inherits:
Object
  • Object
show all
Includes:
Ports::ProcessedTurns
Defined in:
lib/engram/adapters/in_memory_processed_turns.rb

Overview

In-process ProcessedTurns. The zero-config default; guards against double-processing within a single process. For cross-process/retry durability in Rails, use a persistent adapter such as Rails::CacheProcessedTurns.

Instance Method Summary collapse

Constructor Details

#initialize(lease_ttl: 300, clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }) ⇒ InMemoryProcessedTurns

Returns a new instance of InMemoryProcessedTurns.



11
12
13
14
15
16
17
 
# File 'lib/engram/adapters/in_memory_processed_turns.rb', line 11

def initialize(lease_ttl: 300, clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) })
  @lease_ttl = lease_ttl
  @clock = clock
  @claims = {}
  @completed = Set.new
  @mutex = Mutex.new
end

Instance Method Details

#claim(scope:, key:) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/engram/adapters/in_memory_processed_turns.rb', line 19

def claim(scope:, key:)
  scoped_key = [scope, key]
  @mutex.synchronize do
    return if @completed.include?(scoped_key)
    live = @claims[scoped_key]
    return if live && live.last > @clock.call

    token = Object.new
    @claims[scoped_key] = [token, @clock.call + @lease_ttl]
    token
  end
end

#clear ⇒ Object 



53
54
55
56
57
58
 
# File 'lib/engram/adapters/in_memory_processed_turns.rb', line 53

def clear
  @mutex.synchronize do
    @claims.clear
    @completed.clear
  end
end

#complete(scope:, key:, claim:) ⇒ Object 



32
33
34
35
36
37
38
39
40
 
# File 'lib/engram/adapters/in_memory_processed_turns.rb', line 32

def complete(scope:, key:, claim:)
  @mutex.synchronize do
    scoped_key = [scope, key]
    return false unless @claims.dig(scoped_key, 0).equal?(claim)
    @completed << scoped_key
    @claims.delete(scoped_key)
    true
  end
end

#completed?(scope:, key:) ⇒ Boolean

Returns:

  • (Boolean) 


49
50
51
 
# File 'lib/engram/adapters/in_memory_processed_turns.rb', line 49

def completed?(scope:, key:)
  @mutex.synchronize { @completed.include?([scope, key]) }
end

#release(scope:, key:, claim:) ⇒ Object 



42
43
44
45
46
47
 
# File 'lib/engram/adapters/in_memory_processed_turns.rb', line 42

def release(scope:, key:, claim:)
  @mutex.synchronize do
    scoped_key = [scope, key]
    @claims.delete(scoped_key) if @claims.dig(scoped_key, 0).equal?(claim)
  end
end