go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-24 23:02:18 +08:00
Code Issues Projects Releases Wiki Activity

(fix): Requeue to select correct queue

Browse Source
This commit is contained in:
Ken Hibino 2020-01-26 13:41:06 -08:00
parent 31123fd42a
commit 166497748b
2 changed files with 44 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/rdb/rdb.go
View File

@ -156,7 +156,7 @@ func (r *RDB) Requeue(msg *base.TaskMessage) error {
return redis.status_reply("OK") return redis.status_reply("OK")
`) `)
return script.Run(r.client, return script.Run(r.client,
[]string{base.InProgressQueue, base.DefaultQueue}, []string{base.InProgressQueue, base.QueueKey(msg.Queue)},
string(bytes)).Err() string(bytes)).Err()
} }

60
internal/rdb/rdb_test.go
View File

@ -236,33 +236,57 @@ func TestRequeue(t *testing.T) {
r := setup(t) r := setup(t)
t1 := h.NewTaskMessage("send_email", nil) t1 := h.NewTaskMessage("send_email", nil)
t2 := h.NewTaskMessage("export_csv", nil) t2 := h.NewTaskMessage("export_csv", nil)
t3 := h.NewTaskMessageWithQueue("send_email", nil, "critical")
tests := []struct { tests := []struct {
enqueued []*base.TaskMessage // initial state of the default queue enqueued map[string][]*base.TaskMessage // initial state of queues
inProgress []*base.TaskMessage // initial state of the in-progress list inProgress []*base.TaskMessage // initial state of the in-progress list
target *base.TaskMessage // task to requeue target *base.TaskMessage // task to requeue
wantEnqueued []*base.TaskMessage // final state of the default queue wantEnqueued map[string][]*base.TaskMessage // final state of queues
wantInProgress []*base.TaskMessage // final state of the in-progress list wantInProgress []*base.TaskMessage // final state of the in-progress list
}{ }{
{ {
enqueued: []*base.TaskMessage{}, enqueued: map[string][]*base.TaskMessage{
inProgress: []*base.TaskMessage{t1, t2}, base.DefaultQueueName: {},
target: t1, },
wantEnqueued: []*base.TaskMessage{t1}, inProgress: []*base.TaskMessage{t1, t2},
target: t1,
wantEnqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1},
},
wantInProgress: []*base.TaskMessage{t2}, wantInProgress: []*base.TaskMessage{t2},
}, },
{ {
enqueued: []*base.TaskMessage{t1}, enqueued: map[string][]*base.TaskMessage{
inProgress: []*base.TaskMessage{t2}, base.DefaultQueueName: {t1},
target: t2, },
wantEnqueued: []*base.TaskMessage{t1, t2}, inProgress: []*base.TaskMessage{t2},
target: t2,
wantEnqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1, t2},
},
wantInProgress: []*base.TaskMessage{}, wantInProgress: []*base.TaskMessage{},
}, },
{
enqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1},
"critical": {},
},
inProgress: []*base.TaskMessage{t2, t3},
target: t3,
wantEnqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1},
"critical": {t3},
},
wantInProgress: []*base.TaskMessage{t2},
},
} }
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) // clean up db before each test case h.FlushDB(t, r.client) // clean up db before each test case
h.SeedEnqueuedQueue(t, r.client, tc.enqueued) for qname, msgs := range tc.enqueued {
h.SeedEnqueuedQueue(t, r.client, msgs, qname)
}
h.SeedInProgressQueue(t, r.client, tc.inProgress) h.SeedInProgressQueue(t, r.client, tc.inProgress)
err := r.Requeue(tc.target) err := r.Requeue(tc.target)
@ -271,9 +295,11 @@ func TestRequeue(t *testing.T) {
continue continue
} }
gotEnqueued := h.GetEnqueuedMessages(t, r.client) for qname, want := range tc.wantEnqueued {
if diff := cmp.Diff(tc.wantEnqueued, gotEnqueued, h.SortMsgOpt); diff != "" { gotEnqueued := h.GetEnqueuedMessages(t, r.client, qname)
t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.DefaultQueue, diff) if diff := cmp.Diff(want, gotEnqueued, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.QueueKey(qname), diff)
}
} }
gotInProgress := h.GetInProgressMessages(t, r.client) gotInProgress := h.GetInProgressMessages(t, r.client)