apalis-redis 1.0.0

Background task processing for rust using apalis and redis
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

-- KEYS[1]: this worker's inflight set
-- KEYS[2]: the active jobs list
-- KEYS[3]: the signal list
-- KEYS[4]: the active workers sorted set
-- KEYS[5]: the task metadata prefix
-- ARGV[1]: the worker name
-- ARGV[2]: emit event bool ("1" or "0")
-- ARGV[3..]: specific job IDs to release. If empty, release everything in KEYS[1].
-- Returns: number of jobs reenqueued
local worker_inflight_set = KEYS[1]
local worker_name = ARGV[1]

local emit_events = ARGV[2] == "1"

local job_ids = {}
for i = 3, #ARGV do
    table.insert(job_ids, ARGV[i])
end

local reenqueued = 0

if #job_ids > 0 then
    -- Release only the specified jobs
    for _, job_id in ipairs(job_ids) do
        local removed = redis.call("srem", worker_inflight_set, job_id)
        if removed == 1 then
            redis.call("rpush", KEYS[2], job_id)

            local meta_key = KEYS[5] .. ":" .. job_id
            redis.call("hset", meta_key, "status", "Pending")
            redis.call("hdel", meta_key, "locked_at", "locked_by")
            if emit_events then
                redis.call("publish", "tasks:" .. KEYS[2] .. ':available', job_id)
            end
            reenqueued = reenqueued + 1
        end
    end
else
    -- Release everything currently held by this worker
    local all_jobs = redis.call("smembers", worker_inflight_set)
    reenqueued = #all_jobs

    if reenqueued > 0 then
        redis.call("rpush", KEYS[2], unpack(all_jobs))
        redis.call("del", worker_inflight_set)

        for _, job_id in ipairs(all_jobs) do
            local meta_key = KEYS[5] .. ":" .. job_id
            redis.call("hset", meta_key, "status", "Pending")
            redis.call("hdel", meta_key, "locked_at", "locked_by")
            if emit_events then
                redis.call("publish", "tasks:" .. KEYS[2] .. ':available', job_id)
            end
        end
    end
end

if reenqueued > 0 then
    redis.call("del", KEYS[3])
    redis.call("lpush", KEYS[3], 1)
end

return reenqueued