| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116 |
- const content = `--[[
- Attempts to reprocess a job
- Input:
- KEYS[1] job key
- KEYS[2] events stream
- KEYS[3] job state
- KEYS[4] wait key
- KEYS[5] meta
- KEYS[6] paused key
- KEYS[7] active key
- KEYS[8] marker key
- ARGV[1] job.id
- ARGV[2] (job.opts.lifo ? 'R' : 'L') + 'PUSH'
- ARGV[3] propVal - failedReason/returnvalue
- ARGV[4] prev state - failed/completed
- ARGV[5] reset attemptsMade - "1" or "0"
- ARGV[6] reset attemptsStarted - "1" or "0"
- Output:
- 1 means the operation was a success
- -1 means the job does not exist
- -3 means the job was not found in the expected set.
- ]]
- local rcall = redis.call;
- -- Includes
- --[[
- Function to add job in target list and add marker if needed.
- ]]
- -- Includes
- --[[
- Add marker if needed when a job is available.
- ]]
- local function addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)
- if not isPausedOrMaxed then
- rcall("ZADD", markerKey, 0, "0")
- end
- end
- local function addJobInTargetList(targetKey, markerKey, pushCmd, isPausedOrMaxed, jobId)
- rcall(pushCmd, targetKey, jobId)
- addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)
- end
- --[[
- Function to get max events value or set by default 10000.
- ]]
- local function getOrSetMaxEvents(metaKey)
- local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents")
- if not maxEvents then
- maxEvents = 10000
- rcall("HSET", metaKey, "opts.maxLenEvents", maxEvents)
- end
- return maxEvents
- end
- --[[
- Function to check for the meta.paused key to decide if we are paused or not
- (since an empty list and !EXISTS are not really the same).
- ]]
- local function getTargetQueueList(queueMetaKey, activeKey, waitKey, pausedKey)
- local queueAttributes = rcall("HMGET", queueMetaKey, "paused", "concurrency", "max", "duration")
- if queueAttributes[1] then
- return pausedKey, true, queueAttributes[3], queueAttributes[4]
- else
- if queueAttributes[2] then
- local activeCount = rcall("LLEN", activeKey)
- if activeCount >= tonumber(queueAttributes[2]) then
- return waitKey, true, queueAttributes[3], queueAttributes[4]
- else
- return waitKey, false, queueAttributes[3], queueAttributes[4]
- end
- end
- end
- return waitKey, false, queueAttributes[3], queueAttributes[4]
- end
- local jobKey = KEYS[1]
- if rcall("EXISTS", jobKey) == 1 then
- local jobId = ARGV[1]
- if (rcall("ZREM", KEYS[3], jobId) == 1) then
- local attributesToRemove = {}
- if ARGV[5] == "1" then
- table.insert(attributesToRemove, "atm")
- end
- if ARGV[6] == "1" then
- table.insert(attributesToRemove, "ats")
- end
- rcall("HDEL", jobKey, "finishedOn", "processedOn", ARGV[3], unpack(attributesToRemove))
- local target, isPausedOrMaxed = getTargetQueueList(KEYS[5], KEYS[7], KEYS[4], KEYS[6])
- addJobInTargetList(target, KEYS[8], ARGV[2], isPausedOrMaxed, jobId)
- local parentKey = rcall("HGET", jobKey, "parentKey")
- if parentKey and rcall("EXISTS", parentKey) == 1 then
- if ARGV[4] == "failed" then
- if rcall("ZREM", parentKey .. ":unsuccessful", jobKey) == 1 or
- rcall("ZREM", parentKey .. ":failed", jobKey) == 1 then
- rcall("SADD", parentKey .. ":dependencies", jobKey)
- end
- else
- if rcall("HDEL", parentKey .. ":processed", jobKey) == 1 then
- rcall("SADD", parentKey .. ":dependencies", jobKey)
- end
- end
- end
- local maxEvents = getOrSetMaxEvents(KEYS[5])
- -- Emit waiting event
- rcall("XADD", KEYS[2], "MAXLEN", "~", maxEvents, "*", "event", "waiting",
- "jobId", jobId, "prev", ARGV[4]);
- return 1
- else
- return -3
- end
- else
- return -1
- end
- `;
- export const reprocessJob = {
- name: 'reprocessJob',
- content,
- keys: 8,
- };
- //# sourceMappingURL=reprocessJob-8.js.map
|