go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2025-04-22 16:50:18 +08:00
Code Issues Projects Releases Wiki Activity

Code formatting

Browse Source
This commit is contained in:
Ken Hibino 2021-02-22 06:25:40 -08:00
parent 9e7897ae0c
commit b156653243
1 changed files with 41 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
internal/rdb/rdb.go
View File

@ -69,9 +69,15 @@ func (r *RDB) Enqueue(msg *base.TaskMessage) error {
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
} }
keys := []string{base.TaskKey(msg.Queue, msg.ID.String()), base.PendingKey(msg.Queue)} keys := []string{
args := []interface{}{encoded, msg.ID.String()} base.TaskKey(msg.Queue, msg.ID.String()),
return enqueueCmd.Run(r.client, keys, args...).Err() base.PendingKey(msg.Queue),
}
argv := []interface{}{
encoded,
msg.ID.String(),
}
return enqueueCmd.Run(r.client, keys, argv...).Err()
} }
// KEYS[1] -> unique key // KEYS[1] -> unique key
@ -100,9 +106,17 @@ func (r *RDB) EnqueueUnique(msg *base.TaskMessage, ttl time.Duration) error {
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
} }
keys := []string{msg.UniqueKey, base.TaskKey(msg.Queue, msg.ID.String()), base.PendingKey(msg.Queue)} keys := []string{
args := []interface{}{msg.ID.String(), int(ttl.Seconds()), encoded} msg.UniqueKey,
res, err := enqueueUniqueCmd.Run(r.client, keys, args...).Result() base.TaskKey(msg.Queue, msg.ID.String()),
base.PendingKey(msg.Queue),
}
argv := []interface{}{
msg.ID.String(),
int(ttl.Seconds()),
encoded,
}
res, err := enqueueUniqueCmd.Run(r.client, keys, argv...).Result()
if err != nil { if err != nil {
return err return err
} }
@ -308,9 +322,16 @@ func (r *RDB) Schedule(msg *base.TaskMessage, processAt time.Time) error {
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
} }
keys := []string{base.TaskKey(msg.Queue, msg.ID.String()), base.ScheduledKey(msg.Queue)} keys := []string{
args := []interface{}{encoded, processAt.Unix(), msg.ID.String()} base.TaskKey(msg.Queue, msg.ID.String()),
return scheduleCmd.Run(r.client, keys, args...).Err() base.ScheduledKey(msg.Queue),
}
argv := []interface{}{
encoded,
processAt.Unix(),
msg.ID.String(),
}
return scheduleCmd.Run(r.client, keys, argv...).Err()
} }
// KEYS[1] -> unique key // KEYS[1] -> unique key
@ -340,8 +361,17 @@ 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
} }
keys := []string{msg.UniqueKey, base.TaskKey(msg.Queue, msg.ID.String()), base.ScheduledKey(msg.Queue)} keys := []string{
argv := []interface{}{msg.ID.String(), int(ttl.Seconds()), processAt.Unix(), encoded} msg.UniqueKey,
base.TaskKey(msg.Queue, msg.ID.String()),
base.ScheduledKey(msg.Queue),
}
argv := []interface{}{
msg.ID.String(),
int(ttl.Seconds()),
processAt.Unix(),
encoded,
}
res, err := scheduleUniqueCmd.Run(r.client, keys, argv...).Result() res, err := scheduleUniqueCmd.Run(r.client, keys, argv...).Result()
if err != nil { if err != nil {
return err return err