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

Fix inspeq package build

Browse Source
This commit is contained in:
Ken Hibino 2021-02-28 16:00:23 -08:00
parent 6df8c9bed1
commit 7d516e37e6
2 changed files with 16 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
inspeq/inspector.go
View File

@ -548,11 +548,12 @@ func (i *Inspector) DeleteAllArchivedTasks(qname string) (int, error) {
} }
// DeleteTaskByKey deletes a task with the given key from the given queue. // DeleteTaskByKey deletes a task with the given key from the given queue.
// TODO: We don't need score any more. Update this to delete task by ID
func (i *Inspector) DeleteTaskByKey(qname, key string) error { func (i *Inspector) DeleteTaskByKey(qname, key string) error {
if err := base.ValidateQueueName(qname); err != nil { if err := base.ValidateQueueName(qname); err != nil {
return err return err
} }
prefix, id, score, err := parseTaskKey(key) prefix, id, _, err := parseTaskKey(key)
if err != nil { if err != nil {
return err return err
} }
@ -560,11 +561,11 @@ func (i *Inspector) DeleteTaskByKey(qname, key string) error {
case keyPrefixPending: case keyPrefixPending:
return i.rdb.DeletePendingTask(qname, id) return i.rdb.DeletePendingTask(qname, id)
case keyPrefixScheduled: case keyPrefixScheduled:
return i.rdb.DeleteScheduledTask(qname, id, score) return i.rdb.DeleteScheduledTask(qname, id)
case keyPrefixRetry: case keyPrefixRetry:
return i.rdb.DeleteRetryTask(qname, id, score) return i.rdb.DeleteRetryTask(qname, id)
case keyPrefixArchived: case keyPrefixArchived:
return i.rdb.DeleteArchivedTask(qname, id, score) return i.rdb.DeleteArchivedTask(qname, id)
default: default:
return fmt.Errorf("invalid key") return fmt.Errorf("invalid key")
} }
@ -601,21 +602,22 @@ func (i *Inspector) RunAllArchivedTasks(qname string) (int, error) {
} }
// RunTaskByKey transition a task to pending state given task key and queue name. // RunTaskByKey transition a task to pending state given task key and queue name.
// TODO: Update this to run task by ID.
func (i *Inspector) RunTaskByKey(qname, key string) error { func (i *Inspector) RunTaskByKey(qname, key string) error {
if err := base.ValidateQueueName(qname); err != nil { if err := base.ValidateQueueName(qname); err != nil {
return err return err
} }
prefix, id, score, err := parseTaskKey(key) prefix, id, _, err := parseTaskKey(key)
if err != nil { if err != nil {
return err return err
} }
switch prefix { switch prefix {
case keyPrefixScheduled: case keyPrefixScheduled:
return i.rdb.RunScheduledTask(qname, id, score) return i.rdb.RunScheduledTask(qname, id)
case keyPrefixRetry: case keyPrefixRetry:
return i.rdb.RunRetryTask(qname, id, score) return i.rdb.RunRetryTask(qname, id)
case keyPrefixArchived: case keyPrefixArchived:
return i.rdb.RunArchivedTask(qname, id, score) return i.rdb.RunArchivedTask(qname, id)
case keyPrefixPending: case keyPrefixPending:
return fmt.Errorf("task is already pending for run") return fmt.Errorf("task is already pending for run")
default: default:
@ -654,11 +656,12 @@ func (i *Inspector) ArchiveAllRetryTasks(qname string) (int, error) {
} }
// ArchiveTaskByKey archives a task with the given key in the given queue. // ArchiveTaskByKey archives a task with the given key in the given queue.
// TODO: Update this to Archive task by ID.
func (i *Inspector) ArchiveTaskByKey(qname, key string) error { func (i *Inspector) ArchiveTaskByKey(qname, key string) error {
if err := base.ValidateQueueName(qname); err != nil { if err := base.ValidateQueueName(qname); err != nil {
return err return err
} }
prefix, id, score, err := parseTaskKey(key) prefix, id, _, err := parseTaskKey(key)
if err != nil { if err != nil {
return err return err
} }
@ -666,9 +669,9 @@ func (i *Inspector) ArchiveTaskByKey(qname, key string) error {
case keyPrefixPending: case keyPrefixPending:
return i.rdb.ArchivePendingTask(qname, id) return i.rdb.ArchivePendingTask(qname, id)
case keyPrefixScheduled: case keyPrefixScheduled:
return i.rdb.ArchiveScheduledTask(qname, id, score) return i.rdb.ArchiveScheduledTask(qname, id)
case keyPrefixRetry: case keyPrefixRetry:
return i.rdb.ArchiveRetryTask(qname, id, score) return i.rdb.ArchiveRetryTask(qname, id)
case keyPrefixArchived: case keyPrefixArchived:
return fmt.Errorf("task is already archived") return fmt.Errorf("task is already archived")
default: default:

5
inspeq/inspector_test.go
View File

@ -1254,13 +1254,13 @@ func TestInspectorArchiveAllPendingTasks(t *testing.T) {
}, },
{ {
pending: map[string][]*base.TaskMessage{ pending: map[string][]*base.TaskMessage{
"default": {m3, m4}, "default": {m3},
}, },
archived: map[string][]base.Z{ archived: map[string][]base.Z{
"default": {z1, z2}, "default": {z1, z2},
}, },
qname: "default", qname: "default",
want: 2, want: 1,
wantPending: map[string][]*base.TaskMessage{ wantPending: map[string][]*base.TaskMessage{
"default": {}, "default": {},
}, },
@ -1269,7 +1269,6 @@ func TestInspectorArchiveAllPendingTasks(t *testing.T) {
z1, z1,
z2, z2,
base.Z{Message: m3, Score: now.Unix()}, base.Z{Message: m3, Score: now.Unix()},
base.Z{Message: m4, Score: now.Unix()},
}, },
}, },
}, },