| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 |
- const content = `--[[
- Move completed, failed or delayed jobs to wait.
- Note: Does not support jobs with priorities.
- Input:
- KEYS[1] base key
- KEYS[2] events stream
- KEYS[3] state key (failed, completed, delayed)
- KEYS[4] 'wait'
- KEYS[5] 'paused'
- KEYS[6] 'meta'
- KEYS[7] 'active'
- KEYS[8] 'marker'
- ARGV[1] count
- ARGV[2] timestamp
- ARGV[3] prev state
- Output:
- 1 means the operation is not completed
- 0 means the operation is completed
- ]]
- local maxCount = tonumber(ARGV[1])
- local timestamp = tonumber(ARGV[2])
- local rcall = redis.call;
- -- 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
- --[[
- Function to loop in batches.
- Just a bit of warning, some commands as ZREM
- could receive a maximum of 7000 parameters per call.
- ]]
- local function batches(n, batchSize)
- local i = 0
- return function()
- local from = i * batchSize + 1
- i = i + 1
- if (from <= n) then
- local to = math.min(from + batchSize - 1, n)
- return from, to
- end
- end
- 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 metaKey = KEYS[6]
- local target, isPausedOrMaxed = getTargetQueueList(metaKey, KEYS[7], KEYS[4], KEYS[5])
- local jobs = rcall('ZRANGEBYSCORE', KEYS[3], 0, timestamp, 'LIMIT', 0, maxCount)
- if (#jobs > 0) then
- if ARGV[3] == "failed" then
- for i, key in ipairs(jobs) do
- local jobKey = KEYS[1] .. key
- rcall("HDEL", jobKey, "finishedOn", "processedOn", "failedReason")
- end
- elseif ARGV[3] == "completed" then
- for i, key in ipairs(jobs) do
- local jobKey = KEYS[1] .. key
- rcall("HDEL", jobKey, "finishedOn", "processedOn", "returnvalue")
- end
- end
- local maxEvents = getOrSetMaxEvents(metaKey)
- for i, key in ipairs(jobs) do
- -- Emit waiting event
- rcall("XADD", KEYS[2], "MAXLEN", "~", maxEvents, "*", "event",
- "waiting", "jobId", key, "prev", ARGV[3]);
- end
- for from, to in batches(#jobs, 7000) do
- rcall("ZREM", KEYS[3], unpack(jobs, from, to))
- rcall("LPUSH", target, unpack(jobs, from, to))
- end
- addBaseMarkerIfNeeded(KEYS[8], isPausedOrMaxed)
- end
- maxCount = maxCount - #jobs
- if (maxCount <= 0) then return 1 end
- return 0
- `;
- export const moveJobsToWait = {
- name: 'moveJobsToWait',
- content,
- keys: 8,
- };
- //# sourceMappingURL=moveJobsToWait-8.js.map
|