go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2025-07-02 11:23:41 +08:00
Code Issues Projects Releases Wiki Activity

Update RDB.ScheduleUnique

Browse Source
This commit is contained in:
Ken Hibino 2021-02-22 06:21:05 -08:00
parent 69b7aa22cd
commit 9e7897ae0c
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
internal/rdb/rdb.go
View File

@ -314,7 +314,8 @@ func (r *RDB) Schedule(msg *base.TaskMessage, processAt time.Time) error {
} }
// KEYS[1] -> unique key // KEYS[1] -> unique key
// KEYS[2] -> asynq:{<qname>}:scheduled // KEYS[2] -> asynq:{<qname>}:t:<task_id>
// KEYS[3] -> asynq:{<qname>}:scheduled
// ARGV[1] -> task ID // ARGV[1] -> task ID
// ARGV[2] -> uniqueness lock TTL // ARGV[2] -> uniqueness lock TTL
// ARGV[3] -> score (process_at timestamp) // ARGV[3] -> score (process_at timestamp)
@ -324,7 +325,8 @@ local ok = redis.call("SET", KEYS[1], ARGV[1], "NX", "EX", ARGV[2])
if not ok then if not ok then
return 0 return 0
end end
redis.call("ZADD", KEYS[2], ARGV[3], ARGV[4]) redis.call("SET", KEYS[2], ARGV[4])
redis.call("ZADD", KEYS[3], ARGV[3], ARGV[1])
return 1 return 1
`) `)
@ -338,10 +340,9 @@ func (r *RDB) ScheduleUnique(msg *base.TaskMessage, processAt time.Time, ttl tim
if err := r.client.SAdd(base.AllQueues, msg.Queue).Err(); err != nil { if err := r.client.SAdd(base.AllQueues, msg.Queue).Err(); err != nil {
return err return err
} }
score := float64(processAt.Unix()) keys := []string{msg.UniqueKey, base.TaskKey(msg.Queue, msg.ID.String()), base.ScheduledKey(msg.Queue)}
res, err := scheduleUniqueCmd.Run(r.client, argv := []interface{}{msg.ID.String(), int(ttl.Seconds()), processAt.Unix(), encoded}
[]string{msg.UniqueKey, base.ScheduledKey(msg.Queue)}, res, err := scheduleUniqueCmd.Run(r.client, keys, argv...).Result()
msg.ID.String(), int(ttl.Seconds()), score, encoded).Result()
if err != nil { if err != nil {
return err return err
} }