Sha256: b8cd5de63b64fdcc2a6178821cda555b7bda9ccf9f28eed62484ef43706babb0
Contents?: true
Size: 975 Bytes
Versions: 6
Compression:
Stored size: 975 Bytes
Contents
module SidekiqUniqueJobs
module Lock
class WhileExecuting
def self.synchronize(item, redis_pool = nil)
new(item, redis_pool).synchronize { yield }
end
def initialize(item, redis_pool = nil)
@item = item
@mutex = Mutex.new
@redis_pool = redis_pool
@unique_digest = "#{create_digest}:run"
end
def synchronize
@mutex.lock
sleep 0.001 until locked?
yield
ensure
SidekiqUniqueJobs.connection(@redis_pool) { |c| c.del @unique_digest }
@mutex.unlock
end
def locked?
Scripts.call(:synchronize, @redis_pool, keys: [@unique_digest], argv: [Time.now.to_i]) == 1
end
def execute(_callback)
synchronize do
yield
end
end
def create_digest
@unique_digest ||= @item[UNIQUE_DIGEST_KEY]
@unique_digest ||= SidekiqUniqueJobs::UniqueArgs.digest(@item)
end
end
end
end
Version data entries
6 entries across 6 versions & 1 rubygems