-
-
Notifications
You must be signed in to change notification settings - Fork 273
/
reaper.rb
114 lines (102 loc) · 2.57 KB
/
reaper.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
82
83
84
85
86
87
88
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
# frozen_string_literal: true
module SidekiqUniqueJobs
module Orphans
#
# Class DeleteOrphans provides deletion of orphaned digests
#
# @note this is a much slower version of the lua script but does not crash redis
#
# @author Mikael Henriksson <mikael@mhenrixon.com>
#
class Reaper
include SidekiqUniqueJobs::Connection
include SidekiqUniqueJobs::Script::Caller
include SidekiqUniqueJobs::Logging
include SidekiqUniqueJobs::JSON
require_relative "lua_reaper"
require_relative "ruby_reaper"
require_relative "null_reaper"
#
# @return [Hash<Symbol, SidekiqUniqueJobs::Orphans::Reaper] the current implementation of reapers
REAPERS = {
lua: SidekiqUniqueJobs::Orphans::LuaReaper,
ruby: SidekiqUniqueJobs::Orphans::RubyReaper,
none: SidekiqUniqueJobs::Orphans::NullReaper,
nil => SidekiqUniqueJobs::Orphans::NullReaper,
false => SidekiqUniqueJobs::Orphans::NullReaper,
}.freeze
#
# Execute deletion of orphaned digests
#
# @param [Redis] conn nil a connection to redis
#
# @return [void]
#
def self.call(conn = nil)
return new(conn).call if conn
redis { |rcon| new(rcon).call }
end
#
# @!attribute [r] conn
# @return [Redis] a redis connection
attr_reader :conn
#
# Initialize a new instance of DeleteOrphans
#
# @param [Redis] conn a connection to redis
#
def initialize(conn)
@conn = conn
end
#
# Convenient access to the global configuration
#
#
# @return [SidekiqUniqueJobs::Config]
#
def config
SidekiqUniqueJobs.config
end
#
# The reaper that was configured
#
#
# @return [Symbol]
#
def reaper
config.reaper
end
#
# The configured timeout for the reaper
#
#
# @return [Integer] timeout in seconds
#
def reaper_timeout
config.reaper_timeout
end
#
# The number of locks to reap at a time
#
#
# @return [Integer]
#
def reaper_count
config.reaper_count
end
#
# Delete orphaned digests
#
#
# @return [Integer] the number of reaped locks
#
def call
if (implementation = REAPERS[reaper])
implementation.new(conn).call
else
log_fatal(":#{reaper} is invalid for `SidekiqUnqiueJobs.config.reaper`")
end
end
end
end
end