go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-11-10 11:31:58 +08:00
Code Issues Projects Releases Wiki Activity

Update RDB.Dequeue with task state

Browse Source
This commit is contained in:
Ken Hibino 2021-04-26 07:07:12 -07:00
parent 3cadab55cb
commit f2284be43d
2 changed files with 2 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/asynqtest/asynqtest.go
View File

@ -423,7 +423,7 @@ func getMessagesFromZSetWithScores(tb testing.TB, r redis.UniversalClient,
msg := r.HGet(taskKey, "msg").Val() msg := r.HGet(taskKey, "msg").Val()
res = append(res, base.Z{Message: MustUnmarshal(tb, msg), Score: int64(z.Score)}) res = append(res, base.Z{Message: MustUnmarshal(tb, msg), Score: int64(z.Score)})
if gotState := r.HGet(taskKey, "state").Val(); gotState != state { if gotState := r.HGet(taskKey, "state").Val(); gotState != state {
tb.Errorf("task (id=%q) is in state %q, want %q", taskID, gotState, state) tb.Errorf("task (id=%q) is in %q state, want %q", taskID, gotState, state)
} }
} }
return res return res

1
internal/rdb/rdb.go
View File

@ -175,6 +175,7 @@ if redis.call("EXISTS", KEYS[2]) == 0 then
local id = redis.call("RPOPLPUSH", KEYS[1], KEYS[3]) local id = redis.call("RPOPLPUSH", KEYS[1], KEYS[3])
if id then if id then
local key = ARGV[2] .. id local key = ARGV[2] .. id
redis.call("HSET", key, "state", "active")
local data = redis.call("HMGET", key, "msg", "timeout", "deadline") local data = redis.call("HMGET", key, "msg", "timeout", "deadline")
local msg = data[1] local msg = data[1]
local timeout = tonumber(data[2]) local timeout = tonumber(data[2])