49 lines
1.2 KiB
Lua
49 lines
1.2 KiB
Lua
--[[
|
|
Extend locks for multiple jobs and remove them from the stalled set if successful.
|
|
Return the list of job IDs for which the operation failed.
|
|
|
|
KEYS[1] = stalled key
|
|
|
|
ARGV[1] = baseKey
|
|
ARGV[2] = tokens
|
|
ARGV[3] = jobIds
|
|
ARGV[4] = lockDuration (ms)
|
|
|
|
Output:
|
|
An array of failed job IDs. If empty, all succeeded.
|
|
]]
|
|
local rcall = redis.call
|
|
|
|
local stalledKey = KEYS[1]
|
|
local baseKey = ARGV[1]
|
|
local tokens = cmsgpack.unpack(ARGV[2])
|
|
local jobIds = cmsgpack.unpack(ARGV[3])
|
|
local lockDuration = ARGV[4]
|
|
|
|
local jobCount = #jobIds
|
|
local failedJobs = {}
|
|
|
|
for i = 1, jobCount, 1 do
|
|
local lockKey = baseKey .. jobIds[i] .. ':lock'
|
|
local jobId = jobIds[i]
|
|
local token = tokens[i]
|
|
|
|
local currentToken = rcall("GET", lockKey)
|
|
if currentToken then
|
|
if currentToken == token then
|
|
local setResult = rcall("SET", lockKey, token, "PX", lockDuration)
|
|
if setResult then
|
|
rcall("SREM", stalledKey, jobId)
|
|
else
|
|
table.insert(failedJobs, jobId)
|
|
end
|
|
else
|
|
table.insert(failedJobs, jobId)
|
|
end
|
|
else
|
|
table.insert(failedJobs, jobId)
|
|
end
|
|
end
|
|
|
|
return failedJobs
|