go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-26 07:42:17 +08:00
Code Issues Projects Releases Wiki Activity

Refactor enqueue logic for scheduled tasks

Browse Source
This commit is contained in:
Ken Hibino 2019-11-16 07:39:42 -08:00
parent e362e0daca
commit bda718bcaa
1 changed files with 16 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
asynq.go
View File

@ -51,7 +51,6 @@ func NewClient(opt *RedisOpt) *Client {
// Enqueue pushes a given task to the specified queue. // Enqueue pushes a given task to the specified queue.
func (c *Client) Enqueue(queue string, task *Task, executeAt time.Time) error { func (c *Client) Enqueue(queue string, task *Task, executeAt time.Time) error {
if time.Now().After(executeAt) { if time.Now().After(executeAt) {
fmt.Println("going directly to queue")
bytes, err := json.Marshal(task) bytes, err := json.Marshal(task)
if err != nil { if err != nil {
return err return err
@ -156,24 +155,23 @@ func (w *Workers) pollScheduledTasks() {
continue continue
} }
// TODO(hibiken): Acquire lock for job.ID if w.rdb.ZRem(scheduled, j).Val() > 0 {
pipe := w.rdb.TxPipeline() // Do we need to encode this again?
pipe.ZRem(scheduled, j) // Can we skip this entirely by defining Task field to be a string field?
// Do we need to encode this again? bytes, err := json.Marshal(job.Task)
// Can we skip this entirely by defining Task field to be a string field? if err != nil {
bytes, err := json.Marshal(job.Task) log.Printf("could not marshal job.Task %v: %v\n", job.Task, err)
if err != nil { // TODO(hibiken): Put item that cannot marshal into dead queue.
log.Printf("could not marshal job.Task %v: %v\n", job.Task, err) continue
pipe.Discard() }
continue err = w.rdb.RPush(queuePrefix+job.Queue, string(bytes)).Err()
if err != nil {
log.Printf("command RPUSH %q %s failed: %v",
queuePrefix+job.Queue, string(bytes), err)
// TODO(hibiken): Handle this error properly. Add back to scheduled ZSET?
continue
}
} }
pipe.RPush(queuePrefix+job.Queue, string(bytes))
_, err = pipe.Exec()
if err != nil {
log.Printf("could not execute pipeline: %v\n", err)
continue
}
// TODO(hibiken): Release lock for job.ID
} }
} }
} }