56 lines
2.0 KiB
Lua
56 lines
2.0 KiB
Lua
local tickets_key = KEYS[1]
|
|
local worker_index_key = KEYS[2]
|
|
local owners_key = KEYS[3]
|
|
|
|
local worker_id = ARGV[1]
|
|
local limit = tonumber(ARGV[2])
|
|
|
|
local function reply(status, tickets)
|
|
return cjson.encode({status = status, tickets = tickets or {}})
|
|
end
|
|
|
|
local function decode_table(raw)
|
|
if not raw then
|
|
return nil
|
|
end
|
|
local ok, value = pcall(cjson.decode, raw)
|
|
if not ok or type(value) ~= 'table' then
|
|
return nil
|
|
end
|
|
return value
|
|
end
|
|
|
|
local function valid_ticket(ticket, proxy_id)
|
|
return ticket and ticket.version == 1 and ticket.proxyId == proxy_id and
|
|
ticket.workerId == worker_id and type(ticket.workerIndexKey) == 'string' and
|
|
ticket.workerIndexKey == worker_index_key and tonumber(ticket.assignmentEpoch) and
|
|
tonumber(ticket.assignmentEpoch) > 0 and tonumber(ticket.requiredSnapshotEpoch) and
|
|
tonumber(ticket.requiredSnapshotEpoch) > tonumber(ticket.assignmentEpoch)
|
|
end
|
|
|
|
local function valid_assignment(assignment, ticket)
|
|
return assignment and assignment.version == 1 and assignment.draining == true and
|
|
assignment.proxyId == ticket.proxyId and assignment.workerId == ticket.workerId and
|
|
tonumber(assignment.epoch) == tonumber(ticket.assignmentEpoch)
|
|
end
|
|
|
|
if type(worker_id) ~= 'string' or worker_id == '' or not limit or limit <= 0 then
|
|
return reply('invalid')
|
|
end
|
|
|
|
local scan_limit = limit * 4
|
|
local ids = redis.call('ZRANGE', worker_index_key, 0, scan_limit - 1)
|
|
local tickets = {}
|
|
for _, proxy_id in ipairs(ids) do
|
|
local raw = redis.call('HGET', tickets_key, proxy_id)
|
|
local ticket = decode_table(raw)
|
|
local assignment = decode_table(redis.call('HGET', owners_key, proxy_id))
|
|
if valid_ticket(ticket, proxy_id) and valid_assignment(assignment, ticket) and #tickets < limit then
|
|
tickets[#tickets + 1] = raw
|
|
else
|
|
redis.call('HDEL', tickets_key, proxy_id)
|
|
redis.call('ZREM', worker_index_key, proxy_id)
|
|
end
|
|
end
|
|
return reply('ok', tickets)
|