1 line
10 KiB
Plaintext
1 line
10 KiB
Plaintext
{"file":"E:\\np-dms\\lcbp3\\node_modules\\.pnpm\\bullmq@5.65.0\\node_modules\\bullmq\\dist\\cjs\\scripts\\drain-5.js","mappings":"AAAA,YAAY,CAAC;AACb,MAAM,CAAC,cAAc,CAAC,OAAO,EAAE,YAAY,EAAE,EAAE,KAAK,EAAE,IAAI,EAAE,CAAC,CAAC;AAC9D,OAAO,CAAC,KAAK,GAAG,KAAK,CAAC,CAAC;AACvB,MAAM,OAAO,GAAG;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;CAsQf,CAAC;AACF,OAAO,CAAC,KAAK,GAAG;IACZ,IAAI,EAAE,OAAO;IACb,OAAO;IACP,IAAI,EAAE,CAAC;CACV,CAAC","names":[],"sources":["E:\\np-dms\\lcbp3\\node_modules\\.pnpm\\bullmq@5.65.0\\node_modules\\bullmq\\dist\\cjs\\scripts\\drain-5.js"],"sourcesContent":["\"use strict\";\nObject.defineProperty(exports, \"__esModule\", { value: true });\nexports.drain = void 0;\nconst content = `--[[\n Drains the queue, removes all jobs that are waiting\n or delayed, but not active, completed or failed\n Input:\n KEYS[1] 'wait',\n KEYS[2] 'paused'\n KEYS[3] 'delayed'\n KEYS[4] 'prioritized'\n KEYS[5] 'jobschedulers' (repeat)\n ARGV[1] queue key prefix\n ARGV[2] should clean delayed jobs\n]]\nlocal rcall = redis.call\nlocal queueBaseKey = ARGV[1]\n--[[\n Functions to remove jobs.\n]]\n-- Includes\n--[[\n Function to filter out jobs to ignore from a table.\n]]\nlocal function filterOutJobsToIgnore(jobs, jobsToIgnore)\n local filteredJobs = {}\n for i = 1, #jobs do\n if not jobsToIgnore[jobs[i]] then\n table.insert(filteredJobs, jobs[i])\n end\n end\n return filteredJobs\nend\n--[[\n Functions to remove jobs.\n]]\n-- Includes\n--[[\n Function to remove job.\n]]\n-- Includes\n--[[\n Function to remove deduplication key if needed\n when a job is being removed.\n]]\nlocal function removeDeduplicationKeyIfNeededOnRemoval(prefixKey,\n jobKey, jobId)\n local deduplicationId = rcall(\"HGET\", jobKey, \"deid\")\n if deduplicationId then\n local deduplicationKey = prefixKey .. \"de:\" .. deduplicationId\n local currentJobId = rcall('GET', deduplicationKey)\n if currentJobId and currentJobId == jobId then\n return rcall(\"DEL\", deduplicationKey)\n end\n end\nend\n--[[\n Function to remove job keys.\n]]\nlocal function removeJobKeys(jobKey)\n return rcall(\"DEL\", jobKey, jobKey .. ':logs', jobKey .. ':dependencies',\n jobKey .. ':processed', jobKey .. ':failed', jobKey .. ':unsuccessful')\nend\n--[[\n Check if this job has a parent. If so we will just remove it from\n the parent child list, but if it is the last child we should move the parent to \"wait/paused\"\n which requires code from \"moveToFinished\"\n]]\n-- Includes\n--[[\n Function to add job in target list and add marker if needed.\n]]\n-- Includes\n--[[\n Add marker if needed when a job is available.\n]]\nlocal function addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)\n if not isPausedOrMaxed then\n rcall(\"ZADD\", markerKey, 0, \"0\")\n end \nend\nlocal function addJobInTargetList(targetKey, markerKey, pushCmd, isPausedOrMaxed, jobId)\n rcall(pushCmd, targetKey, jobId)\n addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)\nend\n--[[\n Functions to destructure job key.\n Just a bit of warning, these functions may be a bit slow and affect performance significantly.\n]]\nlocal getJobIdFromKey = function (jobKey)\n return string.match(jobKey, \".*:(.*)\")\nend\nlocal getJobKeyPrefix = function (jobKey, jobId)\n return string.sub(jobKey, 0, #jobKey - #jobId)\nend\n--[[\n Function to check for the meta.paused key to decide if we are paused or not\n (since an empty list and !EXISTS are not really the same).\n]]\nlocal function getTargetQueueList(queueMetaKey, activeKey, waitKey, pausedKey)\n local queueAttributes = rcall(\"HMGET\", queueMetaKey, \"paused\", \"concurrency\", \"max\", \"duration\")\n if queueAttributes[1] then\n return pausedKey, true, queueAttributes[3], queueAttributes[4]\n else\n if queueAttributes[2] then\n local activeCount = rcall(\"LLEN\", activeKey)\n if activeCount >= tonumber(queueAttributes[2]) then\n return waitKey, true, queueAttributes[3], queueAttributes[4]\n else\n return waitKey, false, queueAttributes[3], queueAttributes[4]\n end\n end\n end\n return waitKey, false, queueAttributes[3], queueAttributes[4]\nend\nlocal function _moveParentToWait(parentPrefix, parentId, emitEvent)\n local parentTarget, isPausedOrMaxed = getTargetQueueList(parentPrefix .. \"meta\", parentPrefix .. \"active\",\n parentPrefix .. \"wait\", parentPrefix .. \"paused\")\n addJobInTargetList(parentTarget, parentPrefix .. \"marker\", \"RPUSH\", isPausedOrMaxed, parentId)\n if emitEvent then\n local parentEventStream = parentPrefix .. \"events\"\n rcall(\"XADD\", parentEventStream, \"*\", \"event\", \"waiting\", \"jobId\", parentId, \"prev\", \"waiting-children\")\n end\nend\nlocal function removeParentDependencyKey(jobKey, hard, parentKey, baseKey, debounceId)\n if parentKey then\n local parentDependenciesKey = parentKey .. \":dependencies\"\n local result = rcall(\"SREM\", parentDependenciesKey, jobKey)\n if result > 0 then\n local pendingDependencies = rcall(\"SCARD\", parentDependenciesKey)\n if pendingDependencies == 0 then\n local parentId = getJobIdFromKey(parentKey)\n local parentPrefix = getJobKeyPrefix(parentKey, parentId)\n local numRemovedElements = rcall(\"ZREM\", parentPrefix .. \"waiting-children\", parentId)\n if numRemovedElements == 1 then\n if hard then -- remove parent in same queue\n if parentPrefix == baseKey then\n removeParentDependencyKey(parentKey, hard, nil, baseKey, nil)\n removeJobKeys(parentKey)\n if debounceId then\n rcall(\"DEL\", parentPrefix .. \"de:\" .. debounceId)\n end\n else\n _moveParentToWait(parentPrefix, parentId)\n end\n else\n _moveParentToWait(parentPrefix, parentId, true)\n end\n end\n end\n return true\n end\n else\n local parentAttributes = rcall(\"HMGET\", jobKey, \"parentKey\", \"deid\")\n local missedParentKey = parentAttributes[1]\n if( (type(missedParentKey) == \"string\") and missedParentKey ~= \"\"\n and (rcall(\"EXISTS\", missedParentKey) == 1)) then\n local parentDependenciesKey = missedParentKey .. \":dependencies\"\n local result = rcall(\"SREM\", parentDependenciesKey, jobKey)\n if result > 0 then\n local pendingDependencies = rcall(\"SCARD\", parentDependenciesKey)\n if pendingDependencies == 0 then\n local parentId = getJobIdFromKey(missedParentKey)\n local parentPrefix = getJobKeyPrefix(missedParentKey, parentId)\n local numRemovedElements = rcall(\"ZREM\", parentPrefix .. \"waiting-children\", parentId)\n if numRemovedElements == 1 then\n if hard then\n if parentPrefix == baseKey then\n removeParentDependencyKey(missedParentKey, hard, nil, baseKey, nil)\n removeJobKeys(missedParentKey)\n if parentAttributes[2] then\n rcall(\"DEL\", parentPrefix .. \"de:\" .. parentAttributes[2])\n end\n else\n _moveParentToWait(parentPrefix, parentId)\n end\n else\n _moveParentToWait(parentPrefix, parentId, true)\n end\n end\n end\n return true\n end\n end\n end\n return false\nend\nlocal function removeJob(jobId, hard, baseKey, shouldRemoveDeduplicationKey)\n local jobKey = baseKey .. jobId\n removeParentDependencyKey(jobKey, hard, nil, baseKey)\n if shouldRemoveDeduplicationKey then\n removeDeduplicationKeyIfNeededOnRemoval(baseKey, jobKey, jobId)\n end\n removeJobKeys(jobKey)\nend\nlocal function removeJobs(keys, hard, baseKey, max)\n for i, key in ipairs(keys) do\n removeJob(key, hard, baseKey, true --[[remove debounce key]])\n end\n return max - #keys\nend\nlocal function getListItems(keyName, max)\n return rcall('LRANGE', keyName, 0, max - 1)\nend\nlocal function removeListJobs(keyName, hard, baseKey, max, jobsToIgnore)\n local jobs = getListItems(keyName, max)\n if jobsToIgnore then\n jobs = filterOutJobsToIgnore(jobs, jobsToIgnore)\n end\n local count = removeJobs(jobs, hard, baseKey, max)\n rcall(\"LTRIM\", keyName, #jobs, -1)\n return count\nend\n-- Includes\n--[[\n Function to loop in batches.\n Just a bit of warning, some commands as ZREM\n could receive a maximum of 7000 parameters per call.\n]]\nlocal function batches(n, batchSize)\n local i = 0\n return function()\n local from = i * batchSize + 1\n i = i + 1\n if (from <= n) then\n local to = math.min(from + batchSize - 1, n)\n return from, to\n end\n end\nend\n--[[\n Function to get ZSet items.\n]]\nlocal function getZSetItems(keyName, max)\n return rcall('ZRANGE', keyName, 0, max - 1)\nend\nlocal function removeZSetJobs(keyName, hard, baseKey, max, jobsToIgnore)\n local jobs = getZSetItems(keyName, max)\n if jobsToIgnore then\n jobs = filterOutJobsToIgnore(jobs, jobsToIgnore)\n end\n local count = removeJobs(jobs, hard, baseKey, max)\n if(#jobs > 0) then\n for from, to in batches(#jobs, 7000) do\n rcall(\"ZREM\", keyName, unpack(jobs, from, to))\n end\n end\n return count\nend\n-- We must not remove delayed jobs if they are associated to a job scheduler.\nlocal scheduledJobs = {}\nlocal jobSchedulers = rcall(\"ZRANGE\", KEYS[5], 0, -1, \"WITHSCORES\")\n-- For every job scheduler, get the current delayed job id.\nfor i = 1, #jobSchedulers, 2 do\n local jobSchedulerId = jobSchedulers[i]\n local jobSchedulerMillis = jobSchedulers[i + 1]\n local delayedJobId = \"repeat:\" .. jobSchedulerId .. \":\" .. jobSchedulerMillis\n scheduledJobs[delayedJobId] = true\nend\nremoveListJobs(KEYS[1], true, queueBaseKey, 0, scheduledJobs) -- wait\nremoveListJobs(KEYS[2], true, queueBaseKey, 0, scheduledJobs) -- paused\nif ARGV[2] == \"1\" then\n removeZSetJobs(KEYS[3], true, queueBaseKey, 0, scheduledJobs) -- delayed\nend\nremoveZSetJobs(KEYS[4], true, queueBaseKey, 0, scheduledJobs) -- prioritized\n`;\nexports.drain = {\n name: 'drain',\n content,\n keys: 5,\n};\n//# sourceMappingURL=drain-5.js.map"],"version":3} |