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

Update all kill methods in RDB

Browse Source
This commit is contained in:
Ken Hibino 2020-08-15 06:38:33 -07:00
parent 47e9ba4eba
commit 996a6c0ead
2 changed files with 414 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
internal/rdb/inspect.go
View File

@ -420,12 +420,10 @@ func (r *RDB) removeAndEnqueueAll(zset, qkey string) (int64, error) {
return n, nil return n, nil
} }
/* // KillRetryTask finds a retry task that matches the given id and score from the given queue
// KillRetryTask finds a task that matches the given id and score from retry queue // and kills it. If a task that maches the id and score does not exist, it returns ErrTaskNotFound.
// and moves it to dead queue. If a task that maches the id and score does not exist, func (r *RDB) KillRetryTask(qname string, id uuid.UUID, score int64) error {
// it returns ErrTaskNotFound. n, err := r.removeAndKill(base.RetryKey(qname), base.DeadKey(qname), id.String(), float64(score))
func (r *RDB) KillRetryTask(id uuid.UUID, score int64) error {
n, err := r.removeAndKill(base.RetryQueue, id.String(), float64(score))
if err != nil { if err != nil {
return err return err
} }
@ -435,11 +433,10 @@ func (r *RDB) KillRetryTask(id uuid.UUID, score int64) error {
return nil return nil
} }
// KillScheduledTask finds a task that matches the given id and score from scheduled queue // KillScheduledTask finds a scheduled task that matches the given id and score from the given queue
// and moves it to dead queue. If a task that maches the id and score does not exist, // and kills it. If a task that maches the id and score does not exist, it returns ErrTaskNotFound.
// it returns ErrTaskNotFound. func (r *RDB) KillScheduledTask(qname string, id uuid.UUID, score int64) error {
func (r *RDB) KillScheduledTask(id uuid.UUID, score int64) error { n, err := r.removeAndKill(base.ScheduledKey(qname), base.DeadKey(qname), id.String(), float64(score))
n, err := r.removeAndKill(base.ScheduledQueue, id.String(), float64(score))
if err != nil { if err != nil {
return err return err
} }
@ -449,20 +446,20 @@ func (r *RDB) KillScheduledTask(id uuid.UUID, score int64) error {
return nil return nil
} }
// KillAllRetryTasks moves all tasks from retry queue to dead queue and // KillAllRetryTasks kills all retry tasks from the given queue and
// returns the number of tasks that were moved. // returns the number of tasks that were moved.
func (r *RDB) KillAllRetryTasks() (int64, error) { func (r *RDB) KillAllRetryTasks(qname string) (int64, error) {
return r.removeAndKillAll(base.RetryQueue) return r.removeAndKillAll(base.RetryKey(qname), base.DeadKey(qname))
} }
// KillAllScheduledTasks moves all tasks from scheduled queue to dead queue and // KillAllScheduledTasks kills all scheduled tasks from the given queue and
// returns the number of tasks that were moved. // returns the number of tasks that were moved.
func (r *RDB) KillAllScheduledTasks() (int64, error) { func (r *RDB) KillAllScheduledTasks(qname string) (int64, error) {
return r.removeAndKillAll(base.ScheduledQueue) return r.removeAndKillAll(base.ScheduledKey(qname), base.DeadKey(qname))
} }
// KEYS[1] -> ZSET to move task from (e.g., retry queue) // KEYS[1] -> ZSET to move task from (e.g., retry queue)
// KEYS[2] -> asynq:dead // KEYS[2] -> asynq:{<qname>}:dead
// ARGV[1] -> score of the task to kill // ARGV[1] -> score of the task to kill
// ARGV[2] -> id of the task to kill // ARGV[2] -> id of the task to kill
// ARGV[3] -> current timestamp // ARGV[3] -> current timestamp
@ -482,11 +479,11 @@ for _, msg in ipairs(msgs) do
end end
return 0`) return 0`)
func (r *RDB) removeAndKill(zset, id string, score float64) (int64, error) { func (r *RDB) removeAndKill(src, dst, id string, score float64) (int64, error) {
now := time.Now() now := time.Now()
limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago
res, err := removeAndKillCmd.Run(r.client, res, err := removeAndKillCmd.Run(r.client,
[]string{zset, base.DeadQueue}, []string{src, dst},
score, id, now.Unix(), limit, maxDeadTasks).Result() score, id, now.Unix(), limit, maxDeadTasks).Result()
if err != nil { if err != nil {
return 0, err return 0, err
@ -499,7 +496,7 @@ func (r *RDB) removeAndKill(zset, id string, score float64) (int64, error) {
} }
// KEYS[1] -> ZSET to move task from (e.g., retry queue) // KEYS[1] -> ZSET to move task from (e.g., retry queue)
// KEYS[2] -> asynq:dead // KEYS[2] -> asynq:{<qname>}:dead
// ARGV[1] -> current timestamp // ARGV[1] -> current timestamp
// ARGV[2] -> cutoff timestamp (e.g., 90 days ago) // ARGV[2] -> cutoff timestamp (e.g., 90 days ago)
// ARGV[3] -> max number of tasks in dead queue (e.g., 100) // ARGV[3] -> max number of tasks in dead queue (e.g., 100)
@ -513,10 +510,10 @@ for _, msg in ipairs(msgs) do
end end
return table.getn(msgs)`) return table.getn(msgs)`)
func (r *RDB) removeAndKillAll(zset string) (int64, error) { func (r *RDB) removeAndKillAll(src, dst string) (int64, error) {
now := time.Now() now := time.Now()
limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago
res, err := removeAndKillAllCmd.Run(r.client, []string{zset, base.DeadQueue}, res, err := removeAndKillAllCmd.Run(r.client, []string{src, dst},
now.Unix(), limit, maxDeadTasks).Result() now.Unix(), limit, maxDeadTasks).Result()
if err != nil { if err != nil {
return 0, err return 0, err
@ -528,6 +525,7 @@ func (r *RDB) removeAndKillAll(zset string) (int64, error) {
return n, nil return n, nil
} }
/*
// DeleteDeadTask finds a task that matches the given id and score from dead queue // DeleteDeadTask finds a task that matches the given id and score from dead queue
// and deletes it. If a task that matches the id and score does not exist, // and deletes it. If a task that matches the id and score does not exist,
// it returns ErrTaskNotFound. // it returns ErrTaskNotFound.

562
internal/rdb/inspect_test.go
View File

@ -1599,325 +1599,549 @@ func TestEnqueueAllDeadTasks(t *testing.T) {
} }
} }
/*
func TestKillRetryTask(t *testing.T) { func TestKillRetryTask(t *testing.T) {
r := setup(t) r := setup(t)
m1 := h.NewTaskMessage("send_email", nil) m1 := h.NewTaskMessage("task1", nil)
m2 := h.NewTaskMessage("reindex", nil) m2 := h.NewTaskMessage("task2", nil)
t1 := time.Now().Add(time.Minute) m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
t2 := time.Now().Add(time.Hour) m4 := h.NewTaskMessageWithQueue("task4", nil, "custom")
t1 := time.Now().Add(1 * time.Minute)
t2 := time.Now().Add(1 * time.Hour)
t3 := time.Now().Add(2 * time.Hour)
t4 := time.Now().Add(3 * time.Hour)
tests := []struct { tests := []struct {
retry []base.Z retry map[string][]base.Z
dead []base.Z dead map[string][]base.Z
qname string
id uuid.UUID id uuid.UUID
score int64 score int64
want error want error
wantRetry []base.Z wantRetry map[string][]base.Z
wantDead []base.Z wantDead map[string][]base.Z
}{ }{
{ {
retry: []base.Z{ retry: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {
{Message: m2, Score: t2.Unix()}, {Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
}, },
dead: []base.Z{}, dead: map[string][]base.Z{
"default": {},
},
qname: "default",
id: m1.ID, id: m1.ID,
score: t1.Unix(), score: t1.Unix(),
want: nil, want: nil,
wantRetry: []base.Z{ wantRetry: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
}, },
wantDead: []base.Z{ wantDead: map[string][]base.Z{
{Message: m1, Score: time.Now().Unix()}, "default": {{Message: m1, Score: time.Now().Unix()}},
}, },
}, },
{ {
retry: []base.Z{ retry: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {{Message: m1, Score: t1.Unix()}},
}, },
dead: []base.Z{ dead: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
}, },
qname: "default",
id: m2.ID, id: m2.ID,
score: t2.Unix(), score: t2.Unix(),
want: ErrTaskNotFound, want: ErrTaskNotFound,
wantRetry: []base.Z{ wantRetry: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {{Message: m1, Score: t1.Unix()}},
}, },
wantDead: []base.Z{ wantDead: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
},
},
{
retry: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
{Message: m4, Score: t4.Unix()},
},
},
dead: map[string][]base.Z{
"default": {},
"custom": {},
},
qname: "custom",
id: m3.ID,
score: t3.Unix(),
want: nil,
wantRetry: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m4, Score: t4.Unix()},
},
},
wantDead: map[string][]base.Z{
"default": {},
"custom": {{Message: m3, Score: time.Now().Unix()}},
}, },
}, },
} }
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) h.FlushDB(t, r.client)
h.SeedRetryQueue(t, r.client, tc.retry) h.SeedAllRetryQueues(t, r.client, tc.retry)
h.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
got := r.KillRetryTask(tc.id, tc.score) got := r.KillRetryTask(tc.qname, tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("(*RDB).KillRetryTask(%v, %v) = %v, want %v", t.Errorf("(*RDB).KillRetryTask(%q, %v, %v) = %v, want %v",
tc.id, tc.score, got, tc.want) tc.qname, tc.id, tc.score, got, tc.want)
continue continue
} }
gotRetry := h.GetRetryEntries(t, r.client) for qname, want := range tc.wantRetry {
if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotRetry := h.GetRetryEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotRetry, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.RetryQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.RetryKey(qname), diff)
}
} }
gotDead := h.GetDeadEntries(t, r.client) for qname, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotDead := h.GetDeadEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.DeadQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.DeadKey(qname), diff)
}
} }
} }
} }
func TestKillScheduledTask(t *testing.T) { func TestKillScheduledTask(t *testing.T) {
r := setup(t) r := setup(t)
m1 := h.NewTaskMessage("send_email", nil) m1 := h.NewTaskMessage("task1", nil)
m2 := h.NewTaskMessage("reindex", nil) m2 := h.NewTaskMessage("task2", nil)
t1 := time.Now().Add(time.Minute) m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
t2 := time.Now().Add(time.Hour) m4 := h.NewTaskMessageWithQueue("task4", nil, "custom")
t1 := time.Now().Add(1 * time.Minute)
t2 := time.Now().Add(1 * time.Hour)
t3 := time.Now().Add(2 * time.Hour)
t4 := time.Now().Add(3 * time.Hour)
tests := []struct { tests := []struct {
scheduled []base.Z scheduled map[string][]base.Z
dead []base.Z dead map[string][]base.Z
qname string
id uuid.UUID id uuid.UUID
score int64 score int64
want error want error
wantScheduled []base.Z wantScheduled map[string][]base.Z
wantDead []base.Z wantDead map[string][]base.Z
}{ }{
{ {
scheduled: []base.Z{ scheduled: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {
{Message: m2, Score: t2.Unix()}, {Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
}, },
dead: []base.Z{}, dead: map[string][]base.Z{
"default": {},
},
qname: "default",
id: m1.ID, id: m1.ID,
score: t1.Unix(), score: t1.Unix(),
want: nil, want: nil,
wantScheduled: []base.Z{ wantScheduled: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
}, },
wantDead: []base.Z{ wantDead: map[string][]base.Z{
{Message: m1, Score: time.Now().Unix()}, "default": {{Message: m1, Score: time.Now().Unix()}},
}, },
}, },
{ {
scheduled: []base.Z{ scheduled: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {{Message: m1, Score: t1.Unix()}},
}, },
dead: []base.Z{ dead: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
}, },
qname: "default",
id: m2.ID, id: m2.ID,
score: t2.Unix(), score: t2.Unix(),
want: ErrTaskNotFound, want: ErrTaskNotFound,
wantScheduled: []base.Z{ wantScheduled: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {{Message: m1, Score: t1.Unix()}},
}, },
wantDead: []base.Z{ wantDead: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
},
},
{
scheduled: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
{Message: m4, Score: t4.Unix()},
},
},
dead: map[string][]base.Z{
"default": {},
"custom": {},
},
qname: "custom",
id: m3.ID,
score: t3.Unix(),
want: nil,
wantScheduled: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m4, Score: t4.Unix()},
},
},
wantDead: map[string][]base.Z{
"default": {},
"custom": {{Message: m3, Score: time.Now().Unix()}},
}, },
}, },
} }
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) h.FlushDB(t, r.client)
h.SeedScheduledQueue(t, r.client, tc.scheduled) h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
h.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
got := r.KillScheduledTask(tc.id, tc.score) got := r.KillScheduledTask(tc.qname, tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("(*RDB).KillScheduledTask(%v, %v) = %v, want %v", t.Errorf("(*RDB).KillScheduledTask(%q, %v, %v) = %v, want %v",
tc.id, tc.score, got, tc.want) tc.qname, tc.id, tc.score, got, tc.want)
continue continue
} }
gotScheduled := h.GetScheduledEntries(t, r.client) for qname, want := range tc.wantScheduled {
if diff := cmp.Diff(tc.wantScheduled, gotScheduled, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotScheduled := h.GetScheduledEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotScheduled, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.ScheduledQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.ScheduledKey(qname), diff)
}
} }
gotDead := h.GetDeadEntries(t, r.client) for qname, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotDead := h.GetDeadEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.DeadQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.DeadKey(qname), diff)
}
} }
} }
} }
func TestKillAllRetryTasks(t *testing.T) { func TestKillAllRetryTasks(t *testing.T) {
r := setup(t) r := setup(t)
m1 := h.NewTaskMessage("send_email", nil) m1 := h.NewTaskMessage("task1", nil)
m2 := h.NewTaskMessage("reindex", nil) m2 := h.NewTaskMessage("task2", nil)
t1 := time.Now().Add(time.Minute) m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
t2 := time.Now().Add(time.Hour) m4 := h.NewTaskMessageWithQueue("task4", nil, "custom")
t1 := time.Now().Add(1 * time.Minute)
t2 := time.Now().Add(1 * time.Hour)
t3 := time.Now().Add(2 * time.Hour)
t4 := time.Now().Add(3 * time.Hour)
tests := []struct { tests := []struct {
retry []base.Z retry map[string][]base.Z
dead []base.Z dead map[string][]base.Z
qname string
want int64 want int64
wantRetry []base.Z wantRetry map[string][]base.Z
wantDead []base.Z wantDead map[string][]base.Z
}{ }{
{ {
retry: []base.Z{ retry: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {
{Message: m2, Score: t2.Unix()}, {Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
}, },
dead: []base.Z{}, dead: map[string][]base.Z{
want: 2, "default": {},
wantRetry: []base.Z{}, },
wantDead: []base.Z{ qname: "default",
{Message: m1, Score: time.Now().Unix()}, want: 2,
{Message: m2, Score: time.Now().Unix()}, wantRetry: map[string][]base.Z{
"default": {},
},
wantDead: map[string][]base.Z{
"default": {
{Message: m1, Score: time.Now().Unix()},
{Message: m2, Score: time.Now().Unix()},
},
}, },
}, },
{ {
retry: []base.Z{ retry: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {{Message: m1, Score: t1.Unix()}},
}, },
dead: []base.Z{ dead: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
}, },
want: 1, qname: "default",
wantRetry: []base.Z{}, want: 1,
wantDead: []base.Z{ wantRetry: map[string][]base.Z{
{Message: m1, Score: time.Now().Unix()}, "default": {},
{Message: m2, Score: t2.Unix()}, },
wantDead: map[string][]base.Z{
"default": {
{Message: m1, Score: time.Now().Unix()},
{Message: m2, Score: t2.Unix()},
},
}, },
}, },
{ {
retry: []base.Z{}, retry: map[string][]base.Z{
dead: []base.Z{ "default": {},
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
}, },
want: 0, dead: map[string][]base.Z{
wantRetry: []base.Z{}, "default": {
wantDead: []base.Z{ {Message: m1, Score: t1.Unix()},
{Message: m1, Score: t1.Unix()}, {Message: m2, Score: t2.Unix()},
{Message: m2, Score: t2.Unix()}, },
},
qname: "default",
want: 0,
wantRetry: map[string][]base.Z{
"default": {},
},
wantDead: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
},
},
{
retry: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
{Message: m4, Score: t4.Unix()},
},
},
dead: map[string][]base.Z{
"default": {},
"custom": {},
},
qname: "custom",
want: 2,
wantRetry: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {},
},
wantDead: map[string][]base.Z{
"default": {},
"custom": {
{Message: m3, Score: time.Now().Unix()},
{Message: m4, Score: time.Now().Unix()},
},
}, },
}, },
} }
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) h.FlushDB(t, r.client)
h.SeedRetryQueue(t, r.client, tc.retry) h.SeedAllRetryQueues(t, r.client, tc.retry)
h.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
got, err := r.KillAllRetryTasks() got, err := r.KillAllRetryTasks(tc.qname)
if got != tc.want || err != nil { if got != tc.want || err != nil {
t.Errorf("(*RDB).KillAllRetryTasks() = %v, %v; want %v, nil", t.Errorf("(*RDB).KillAllRetryTasks(%q) = %v, %v; want %v, nil",
got, err, tc.want) tc.qname, got, err, tc.want)
continue continue
} }
gotRetry := h.GetRetryEntries(t, r.client) for qname, want := range tc.wantRetry {
if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotRetry := h.GetRetryEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotRetry, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.RetryQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.RetryKey(qname), diff)
}
} }
gotDead := h.GetDeadEntries(t, r.client) for qname, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotDead := h.GetDeadEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.DeadQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.DeadKey(qname), diff)
}
} }
} }
} }
func TestKillAllScheduledTasks(t *testing.T) { func TestKillAllScheduledTasks(t *testing.T) {
r := setup(t) r := setup(t)
m1 := h.NewTaskMessage("send_email", nil) m1 := h.NewTaskMessage("task1", nil)
m2 := h.NewTaskMessage("reindex", nil) m2 := h.NewTaskMessage("task2", nil)
m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
m4 := h.NewTaskMessageWithQueue("task4", nil, "custom")
t1 := time.Now().Add(time.Minute) t1 := time.Now().Add(time.Minute)
t2 := time.Now().Add(time.Hour) t2 := time.Now().Add(time.Hour)
t3 := time.Now().Add(time.Hour)
t4 := time.Now().Add(time.Hour)
tests := []struct { tests := []struct {
scheduled []base.Z scheduled map[string][]base.Z
dead []base.Z dead map[string][]base.Z
qname string
want int64 want int64
wantScheduled []base.Z wantScheduled map[string][]base.Z
wantDead []base.Z wantDead map[string][]base.Z
}{ }{
{ {
scheduled: []base.Z{ scheduled: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {
{Message: m2, Score: t2.Unix()}, {Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
}, },
dead: []base.Z{}, dead: map[string][]base.Z{
want: 2, "default": {},
wantScheduled: []base.Z{}, },
wantDead: []base.Z{ qname: "default",
{Message: m1, Score: time.Now().Unix()}, want: 2,
{Message: m2, Score: time.Now().Unix()}, wantScheduled: map[string][]base.Z{
"default": {},
},
wantDead: map[string][]base.Z{
"default": {
{Message: m1, Score: time.Now().Unix()},
{Message: m2, Score: time.Now().Unix()},
},
}, },
}, },
{ {
scheduled: []base.Z{ scheduled: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {{Message: m1, Score: t1.Unix()}},
}, },
dead: []base.Z{ dead: map[string][]base.Z{
{Message: m2, Score: t2.Unix()}, "default": {{Message: m2, Score: t2.Unix()}},
}, },
want: 1, qname: "default",
wantScheduled: []base.Z{}, want: 1,
wantDead: []base.Z{ wantScheduled: map[string][]base.Z{
{Message: m1, Score: time.Now().Unix()}, "default": {},
{Message: m2, Score: t2.Unix()}, },
wantDead: map[string][]base.Z{
"default": {
{Message: m1, Score: time.Now().Unix()},
{Message: m2, Score: t2.Unix()},
},
}, },
}, },
{ {
scheduled: []base.Z{}, scheduled: map[string][]base.Z{
dead: []base.Z{ "default": {},
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
}, },
want: 0, dead: map[string][]base.Z{
wantScheduled: []base.Z{}, "default": {
wantDead: []base.Z{ {Message: m1, Score: t1.Unix()},
{Message: m1, Score: t1.Unix()}, {Message: m2, Score: t2.Unix()},
{Message: m2, Score: t2.Unix()}, },
},
qname: "default",
want: 0,
wantScheduled: map[string][]base.Z{
"default": {},
},
wantDead: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
},
},
{
scheduled: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
{Message: m4, Score: t4.Unix()},
},
},
dead: map[string][]base.Z{
"default": {},
"custom": {},
},
qname: "custom",
want: 2,
wantScheduled: map[string][]base.Z{
"default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {},
},
wantDead: map[string][]base.Z{
"default": {},
"custom": {
{Message: m3, Score: time.Now().Unix()},
{Message: m4, Score: time.Now().Unix()},
},
}, },
}, },
} }
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) h.FlushDB(t, r.client)
h.SeedScheduledQueue(t, r.client, tc.scheduled) h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
h.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
got, err := r.KillAllScheduledTasks() got, err := r.KillAllScheduledTasks(tc.qname)
if got != tc.want || err != nil { if got != tc.want || err != nil {
t.Errorf("(*RDB).KillAllScheduledTasks() = %v, %v; want %v, nil", t.Errorf("(*RDB).KillAllScheduledTasks(%q) = %v, %v; want %v, nil",
got, err, tc.want) tc.qname, got, err, tc.want)
continue continue
} }
gotScheduled := h.GetScheduledEntries(t, r.client) for qname, want := range tc.wantScheduled {
if diff := cmp.Diff(tc.wantScheduled, gotScheduled, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotScheduled := h.GetScheduledEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotScheduled, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.ScheduledQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.ScheduledKey(qname), diff)
}
} }
gotDead := h.GetDeadEntries(t, r.client) for qname, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" { gotDead := h.GetDeadEntries(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want,+got)\n%s", if diff := cmp.Diff(want, gotDead, h.SortZSetEntryOpt, timeCmpOpt); diff != "" {
base.DeadQueue, diff) t.Errorf("mismatch found in %q; (-want,+got)\n%s",
base.DeadKey(qname), diff)
}
} }
} }
} }
/*
func TestDeleteDeadTask(t *testing.T) { func TestDeleteDeadTask(t *testing.T) {
r := setup(t) r := setup(t)
m1 := h.NewTaskMessage("send_email", nil) m1 := h.NewTaskMessage("send_email", nil)