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

Rename NextEnqueueAt to NextProcessAt

Browse Source
This commit is contained in:
Ken Hibino 2020-09-05 07:11:55 -07:00
parent c8c47fcbf0
commit ebe3c4083f
3 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
inspector.go
View File

@ -138,7 +138,7 @@ type ScheduledTask struct {
*Task *Task
ID string ID string
Queue string Queue string
NextEnqueueAt time.Time NextProcessAt time.Time
score int64 score int64
} }
@ -148,7 +148,7 @@ type RetryTask struct {
*Task *Task
ID string ID string
Queue string Queue string
NextEnqueueAt time.Time NextProcessAt time.Time
MaxRetry int MaxRetry int
Retried int Retried int
ErrorMsg string ErrorMsg string
@ -318,7 +318,7 @@ func (i *Inspector) ListInProgressTasks(qname string, opts ...ListOption) ([]*In
} }
// ListScheduledTasks retrieves scheduled tasks from the specified queue. // ListScheduledTasks retrieves scheduled tasks from the specified queue.
// Tasks are sorted by NextEnqueueAt field in ascending order. // Tasks are sorted by NextProcessAt field in ascending order.
// //
// By default, it retrieves the first 30 tasks. // By default, it retrieves the first 30 tasks.
func (i *Inspector) ListScheduledTasks(qname string, opts ...ListOption) ([]*ScheduledTask, error) { func (i *Inspector) ListScheduledTasks(qname string, opts ...ListOption) ([]*ScheduledTask, error) {
@ -333,13 +333,13 @@ func (i *Inspector) ListScheduledTasks(qname string, opts ...ListOption) ([]*Sch
} }
var tasks []*ScheduledTask var tasks []*ScheduledTask
for _, z := range zs { for _, z := range zs {
enqueueAt := time.Unix(z.Score, 0) processAt := time.Unix(z.Score, 0)
t := NewTask(z.Message.Type, z.Message.Payload) t := NewTask(z.Message.Type, z.Message.Payload)
tasks = append(tasks, &ScheduledTask{ tasks = append(tasks, &ScheduledTask{
Task: t, Task: t,
ID: z.Message.ID.String(), ID: z.Message.ID.String(),
Queue: z.Message.Queue, Queue: z.Message.Queue,
NextEnqueueAt: enqueueAt, NextProcessAt: processAt,
score: z.Score, score: z.Score,
}) })
} }
@ -347,7 +347,7 @@ func (i *Inspector) ListScheduledTasks(qname string, opts ...ListOption) ([]*Sch
} }
// ListRetryTasks retrieves retry tasks from the specified queue. // ListRetryTasks retrieves retry tasks from the specified queue.
// Tasks are sorted by NextEnqueueAt field in ascending order. // Tasks are sorted by NextProcessAt field in ascending order.
// //
// By default, it retrieves the first 30 tasks. // By default, it retrieves the first 30 tasks.
func (i *Inspector) ListRetryTasks(qname string, opts ...ListOption) ([]*RetryTask, error) { func (i *Inspector) ListRetryTasks(qname string, opts ...ListOption) ([]*RetryTask, error) {
@ -362,13 +362,13 @@ func (i *Inspector) ListRetryTasks(qname string, opts ...ListOption) ([]*RetryTa
} }
var tasks []*RetryTask var tasks []*RetryTask
for _, z := range zs { for _, z := range zs {
enqueueAt := time.Unix(z.Score, 0) processAt := time.Unix(z.Score, 0)
t := NewTask(z.Message.Type, z.Message.Payload) t := NewTask(z.Message.Type, z.Message.Payload)
tasks = append(tasks, &RetryTask{ tasks = append(tasks, &RetryTask{
Task: t, Task: t,
ID: z.Message.ID.String(), ID: z.Message.ID.String(),
Queue: z.Message.Queue, Queue: z.Message.Queue,
NextEnqueueAt: enqueueAt, NextProcessAt: processAt,
MaxRetry: z.Message.Retry, MaxRetry: z.Message.Retry,
Retried: z.Message.Retried, Retried: z.Message.Retried,
// TODO: LastFailedAt: z.Message.LastFailedAt // TODO: LastFailedAt: z.Message.LastFailedAt

8
inspector_test.go
View File

@ -351,7 +351,7 @@ func createScheduledTask(z base.Z) *ScheduledTask {
Task: NewTask(msg.Type, msg.Payload), Task: NewTask(msg.Type, msg.Payload),
ID: msg.ID.String(), ID: msg.ID.String(),
Queue: msg.Queue, Queue: msg.Queue,
NextEnqueueAt: time.Unix(z.Score, 0), NextProcessAt: time.Unix(z.Score, 0),
score: z.Score, score: z.Score,
} }
} }
@ -383,7 +383,7 @@ func TestInspectorListScheduledTasks(t *testing.T) {
"custom": {z4}, "custom": {z4},
}, },
qname: "default", qname: "default",
// Should be sorted by NextEnqueuedAt. // Should be sorted by NextProcessAt.
want: []*ScheduledTask{ want: []*ScheduledTask{
createScheduledTask(z3), createScheduledTask(z3),
createScheduledTask(z1), createScheduledTask(z1),
@ -423,7 +423,7 @@ func createRetryTask(z base.Z) *RetryTask {
Task: NewTask(msg.Type, msg.Payload), Task: NewTask(msg.Type, msg.Payload),
ID: msg.ID.String(), ID: msg.ID.String(),
Queue: msg.Queue, Queue: msg.Queue,
NextEnqueueAt: time.Unix(z.Score, 0), NextProcessAt: time.Unix(z.Score, 0),
MaxRetry: msg.Retry, MaxRetry: msg.Retry,
Retried: msg.Retried, Retried: msg.Retried,
ErrorMsg: msg.ErrorMsg, ErrorMsg: msg.ErrorMsg,
@ -458,7 +458,7 @@ func TestInspectorListRetryTasks(t *testing.T) {
"custom": {z4}, "custom": {z4},
}, },
qname: "default", qname: "default",
// Should be sorted by NextEnqueuedAt. // Should be sorted by NextProcessAt.
want: []*RetryTask{ want: []*RetryTask{
createRetryTask(z3), createRetryTask(z3),
createRetryTask(z1), createRetryTask(z1),

4
tools/asynq/cmd/task.go
View File

@ -239,7 +239,7 @@ func listScheduledTasks(qname string, pageNum, pageSize int) {
func(w io.Writer, tmpl string) { func(w io.Writer, tmpl string) {
for _, t := range tasks { for _, t := range tasks {
processIn := fmt.Sprintf("%.0f seconds", processIn := fmt.Sprintf("%.0f seconds",
t.NextEnqueueAt.Sub(time.Now()).Seconds()) t.NextProcessAt.Sub(time.Now()).Seconds())
fmt.Fprintf(w, tmpl, t.Key(), t.Type, t.Payload, processIn) fmt.Fprintf(w, tmpl, t.Key(), t.Type, t.Payload, processIn)
} }
}, },
@ -262,7 +262,7 @@ func listRetryTasks(qname string, pageNum, pageSize int) {
func(w io.Writer, tmpl string) { func(w io.Writer, tmpl string) {
for _, t := range tasks { for _, t := range tasks {
var nextRetry string var nextRetry string
if d := t.NextEnqueueAt.Sub(time.Now()); d > 0 { if d := t.NextProcessAt.Sub(time.Now()); d > 0 {
nextRetry = fmt.Sprintf("in %v", d.Round(time.Second)) nextRetry = fmt.Sprintf("in %v", d.Round(time.Second))
} else { } else {
nextRetry = "right now" nextRetry = "right now"