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

Update Requeue method in RDB

Browse Source
This commit is contained in:
Ken Hibino 2020-08-09 05:40:44 -07:00
parent 74f08795f8
commit f6d504939e
2 changed files with 69 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/rdb/rdb.go
View File

@ -226,9 +226,9 @@ func (r *RDB) Done(msg *base.TaskMessage) error {
encoded, expireAt.Unix(), msg.ID.String()).Err() encoded, expireAt.Unix(), msg.ID.String()).Err()
} }
// KEYS[1] -> asynq:in_progress // KEYS[1] -> asynq:{<qname>}:in_progress
// KEYS[2] -> asynq:deadlines // KEYS[2] -> asynq:{<qname>}:deadlines
// KEYS[3] -> asynq:queues:<qname> // KEYS[3] -> asynq:{<qname>}
// ARGV[1] -> base.TaskMessage value // ARGV[1] -> base.TaskMessage value
// Note: Use RPUSH to push to the head of the queue. // Note: Use RPUSH to push to the head of the queue.
var requeueCmd = redis.NewScript(` var requeueCmd = redis.NewScript(`
@ -248,7 +248,7 @@ func (r *RDB) Requeue(msg *base.TaskMessage) error {
return err return err
} }
return requeueCmd.Run(r.client, return requeueCmd.Run(r.client,
[]string{base.InProgressQueue, base.KeyDeadlines, base.QueueKey(msg.Queue)}, []string{base.InProgressKey(msg.Queue), base.DeadlinesKey(msg.Queue), base.QueueKey(msg.Queue)},
encoded).Err() encoded).Err()
} }

106
internal/rdb/rdb_test.go
View File

@ -583,75 +583,96 @@ func TestRequeue(t *testing.T) {
tests := []struct { tests := []struct {
enqueued map[string][]*base.TaskMessage // initial state of queues enqueued map[string][]*base.TaskMessage // initial state of queues
inProgress []*base.TaskMessage // initial state of the in-progress list inProgress map[string][]*base.TaskMessage // initial state of the in-progress list
deadlines []base.Z // initial state of the deadlines set deadlines map[string][]base.Z // initial state of the deadlines set
target *base.TaskMessage // task to requeue target *base.TaskMessage // task to requeue
wantEnqueued map[string][]*base.TaskMessage // final state of queues wantEnqueued map[string][]*base.TaskMessage // final state of queues
wantInProgress []*base.TaskMessage // final state of the in-progress list wantInProgress map[string][]*base.TaskMessage // final state of the in-progress list
wantDeadlines []base.Z // final state of the deadlines set wantDeadlines map[string][]base.Z // final state of the deadlines set
}{ }{
{ {
enqueued: map[string][]*base.TaskMessage{ enqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {}, "default": {},
},
inProgress: map[string][]*base.TaskMessage{
"default": {t1, t2},
}, },
inProgress: []*base.TaskMessage{t1, t2},
deadlines: []base.Z{ deadlines: []base.Z{
{Message: t1, Score: t1Deadline}, "default": {
{Message: t2, Score: t2Deadline}, {Message: t1, Score: t1Deadline},
{Message: t2, Score: t2Deadline},
},
}, },
target: t1, target: t1,
wantEnqueued: map[string][]*base.TaskMessage{ wantEnqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1}, "default": {t1},
},
wantInProgress: map[string][]*base.TaskMessage{
"default": {t2},
}, },
wantInProgress: []*base.TaskMessage{t2},
wantDeadlines: []base.Z{ wantDeadlines: []base.Z{
{Message: t2, Score: t2Deadline}, "defult": {
{Message: t2, Score: t2Deadline},
},
}, },
}, },
{ {
enqueued: map[string][]*base.TaskMessage{ enqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1}, "default": {t1},
}, },
inProgress: []*base.TaskMessage{t2}, inProgress: map[string][]*base.TaskMessage{
deadlines: []base.Z{ "default": {t2},
{Message: t2, Score: t2Deadline}, },
deadlines: map[string][]base.Z{
"default": {
{Message: t2, Score: t2Deadline},
},
}, },
target: t2, target: t2,
wantEnqueued: map[string][]*base.TaskMessage{ wantEnqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1, t2}, "default": {t1, t2},
},
wantInProgress: map[string][]*base.TaskMessage{
"default": {},
},
wantDeadlines: map[string][]base.Z{
"default": {},
}, },
wantInProgress: []*base.TaskMessage{},
wantDeadlines: []base.Z{},
}, },
{ {
enqueued: map[string][]*base.TaskMessage{ enqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1}, "default": {t1},
"critical": {}, "critical": {},
}, },
inProgress: []*base.TaskMessage{t2, t3}, inProgress: map[string][]*base.TaskMessage{
deadlines: []base.Z{ "default": {t2},
{Message: t2, Score: t2Deadline}, "critical": {t3},
{Message: t3, Score: t3Deadline}, },
deadlines: map[string][]base.Z{
"defualt": {{Message: t2, Score: t2Deadline}},
"critial": {{Message: t3, Score: t3Deadline}},
}, },
target: t3, target: t3,
wantEnqueued: map[string][]*base.TaskMessage{ wantEnqueued: map[string][]*base.TaskMessage{
base.DefaultQueueName: {t1}, "default": {t1},
"critical": {t3}, "critical": {t3},
}, },
wantInProgress: []*base.TaskMessage{t2}, wantInProgress: map[string][]*base.TaskMessage{
wantDeadlines: []base.Z{ "default": {t2},
{Message: t2, Score: t2Deadline}, "critical": {},
},
wantDeadlines: map[string][]base.Z{
"default": {{Message: t2, Score: t2Deadline}},
"critical": {},
}, },
}, },
} }
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
for qname, msgs := range tc.enqueued { h.SeedAllEnqueuedQueues(t, r.client, msgs, tc.enqueued)
h.SeedEnqueuedQueue(t, r.client, msgs, qname) h.SeedAllInProgressQueues(t, r.client, tc.inProgress)
} h.SeedAllDeadlines(t, r.client, tc.deadlines)
h.SeedInProgressQueue(t, r.client, tc.inProgress)
h.SeedDeadlines(t, r.client, tc.deadlines)
err := r.Requeue(tc.target) err := r.Requeue(tc.target)
if err != nil { if err != nil {
@ -665,14 +686,17 @@ func TestRequeue(t *testing.T) {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.QueueKey(qname), diff) t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.QueueKey(qname), diff)
} }
} }
for qname, want := range tc.wantInProgress {
gotInProgress := h.GetInProgressMessages(t, r.client) gotInProgress := h.GetInProgressMessages(t, r.client, qname)
if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" { if diff := cmp.Diff(want, gotInProgress, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.InProgressQueue, diff) t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.InProgressKey(qname), diff)
}
} }
gotDeadlines := h.GetDeadlinesEntries(t, r.client) for qname, want := range tc.wantDeadlines {
if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" { gotDeadlines := h.GetDeadlinesEntries(t, r.client, qname)
t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.KeyDeadlines, diff) if diff := cmp.Diff(wnt, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.DeadlinesKey(qname), diff)
}
} }
} }
} }