go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-09-20 11:05:58 +08:00
Code Issues Projects Releases Wiki Activity

Update all delete methods in RDB

Browse Source
This commit is contained in:
Ken Hibino 2020-08-15 13:04:26 -07:00
parent 996a6c0ead
commit f59de9ac56
2 changed files with 313 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
internal/rdb/inspect.go
View File

@ -525,26 +525,22 @@ func (r *RDB) removeAndKillAll(src, dst string) (int64, error) {
return n, nil return n, nil
} }
/* // DeleteDeadTask deletes a dead task that matches the given id and score from the given queue.
// DeleteDeadTask finds a task that matches the given id and score from dead queue // If a task that matches the id and score does not exist, it returns ErrTaskNotFound.
// and deletes it. If a task that matches the id and score does not exist, func (r *RDB) DeleteDeadTask(qname string, id uuid.UUID, score int64) error {
// it returns ErrTaskNotFound. return r.deleteTask(base.DeadKey(qname), id.String(), float64(score))
func (r *RDB) DeleteDeadTask(id uuid.UUID, score int64) error {
return r.deleteTask(base.DeadQueue, id.String(), float64(score))
} }
// DeleteRetryTask finds a task that matches the given id and score from retry queue // DeleteRetryTask deletes a retry task that matches the given id and score from the given queue.
// and deletes it. If a task that matches the id and score does not exist, // If a task that matches the id and score does not exist, it returns ErrTaskNotFound.
// it returns ErrTaskNotFound. func (r *RDB) DeleteRetryTask(qname string, id uuid.UUID, score int64) error {
func (r *RDB) DeleteRetryTask(id uuid.UUID, score int64) error { return r.deleteTask(base.RetryKey(qname), id.String(), float64(score))
return r.deleteTask(base.RetryQueue, id.String(), float64(score))
} }
// DeleteScheduledTask finds a task that matches the given id and score from // DeleteScheduledTask deletes a scheduled task that matches the given id and score from the given queue.
// scheduled queue and deletes it. If a task that matches the id and score // If a task that matches the id and score does not exist, it returns ErrTaskNotFound.
//does not exist, it returns ErrTaskNotFound. func (r *RDB) DeleteScheduledTask(qname string, id uuid.UUID, score int64) error {
func (r *RDB) DeleteScheduledTask(id uuid.UUID, score int64) error { return r.deleteTask(base.ScheduledKey(qname), id.String(), float64(score))
return r.deleteTask(base.ScheduledQueue, id.String(), float64(score))
} }
var deleteTaskCmd = redis.NewScript(` var deleteTaskCmd = redis.NewScript(`
@ -558,8 +554,8 @@ for _, msg in ipairs(msgs) do
end end
return 0`) return 0`)
func (r *RDB) deleteTask(zset, id string, score float64) error { func (r *RDB) deleteTask(key, id string, score float64) error {
res, err := deleteTaskCmd.Run(r.client, []string{zset}, score, id).Result() res, err := deleteTaskCmd.Run(r.client, []string{key}, score, id).Result()
if err != nil { if err != nil {
return err return err
} }
@ -579,22 +575,22 @@ local n = redis.call("ZCARD", KEYS[1])
redis.call("DEL", KEYS[1]) redis.call("DEL", KEYS[1])
return n`) return n`)
// DeleteAllDeadTasks deletes all tasks from the dead queue // DeleteAllDeadTasks deletes all dead tasks from the given queue
// and returns the number of tasks deleted. // and returns the number of tasks deleted.
func (r *RDB) DeleteAllDeadTasks() (int64, error) { func (r *RDB) DeleteAllDeadTasks(qname string) (int64, error) {
return r.deleteAll(base.DeadQueue) return r.deleteAll(base.DeadKey(qname))
} }
// DeleteAllRetryTasks deletes all tasks from the dead queue // DeleteAllRetryTasks deletes all retry tasks from the given queue
// and returns the number of tasks deleted. // and returns the number of tasks deleted.
func (r *RDB) DeleteAllRetryTasks() (int64, error) { func (r *RDB) DeleteAllRetryTasks(qname string) (int64, error) {
return r.deleteAll(base.RetryQueue) return r.deleteAll(base.RetryKey(qname))
} }
// DeleteAllScheduledTasks deletes all tasks from the dead queue // DeleteAllScheduledTasks deletes all scheduled tasks from the given queue
// and returns the number of tasks deleted. // and returns the number of tasks deleted.
func (r *RDB) DeleteAllScheduledTasks() (int64, error) { func (r *RDB) DeleteAllScheduledTasks(qname string) (int64, error) {
return r.deleteAll(base.ScheduledQueue) return r.deleteAll(base.ScheduledKey(qname))
} }
func (r *RDB) deleteAll(key string) (int64, error) { func (r *RDB) deleteAll(key string) (int64, error) {
@ -608,7 +604,6 @@ func (r *RDB) deleteAll(key string) (int64, error) {
} }
return n, nil return n, nil
} }
*/
// ErrQueueNotFound indicates specified queue does not exist. // ErrQueueNotFound indicates specified queue does not exist.
type ErrQueueNotFound struct { type ErrQueueNotFound struct {

429
internal/rdb/inspect_test.go
View File

@ -2141,306 +2141,457 @@ func TestKillAllScheduledTasks(t *testing.T) {
} }
} }
/*
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("task1", nil)
m2 := h.NewTaskMessage("reindex", nil) m2 := h.NewTaskMessage("task2", nil)
m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
t1 := time.Now().Add(-5 * time.Minute) t1 := time.Now().Add(-5 * time.Minute)
t2 := time.Now().Add(-time.Hour) t2 := time.Now().Add(-time.Hour)
t3 := time.Now().Add(-time.Hour)
tests := []struct { tests := []struct {
dead []base.Z dead map[string][]base.Z
qname string
id uuid.UUID id uuid.UUID
score int64 score int64
want error want error
wantDead []*base.TaskMessage wantDead map[string][]*base.TaskMessage
}{ }{
{ {
dead: []base.Z{ dead: 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()},
},
},
qname: "default",
id: m1.ID,
score: t1.Unix(),
want: nil,
wantDead: map[string][]*base.TaskMessage{
"default": {m2},
}, },
id: m1.ID,
score: t1.Unix(),
want: nil,
wantDead: []*base.TaskMessage{m2},
}, },
{ {
dead: []base.Z{ dead: 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()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
},
},
qname: "custom",
id: m3.ID,
score: t3.Unix(),
want: nil,
wantDead: map[string][]*base.TaskMessage{
"default": {m1, m2},
"custom": {},
}, },
id: m1.ID,
score: t2.Unix(), // id and score mismatch
want: ErrTaskNotFound,
wantDead: []*base.TaskMessage{m1, m2},
}, },
{ {
dead: []base.Z{}, dead: map[string][]base.Z{
id: m1.ID, "default": {
score: t1.Unix(), {Message: m1, Score: t1.Unix()},
want: ErrTaskNotFound, {Message: m2, Score: t2.Unix()},
wantDead: []*base.TaskMessage{}, },
},
qname: "default",
id: m1.ID,
score: t2.Unix(), // id and score mismatch
want: ErrTaskNotFound,
wantDead: map[string][]*base.TaskMessage{
"default": {m1, m2},
},
},
{
dead: map[string][]base.Z{
"default": {},
},
qname: "default",
id: m1.ID,
score: t1.Unix(),
want: ErrTaskNotFound,
wantDead: map[string][]*base.TaskMessage{
"default": {},
},
}, },
} }
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.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
got := r.DeleteDeadTask(tc.id, tc.score) got := r.DeleteDeadTask(tc.qname, tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("r.DeleteDeadTask(%v, %v) = %v, want %v", tc.id, tc.score, got, tc.want) t.Errorf("r.DeleteDeadTask(%q, %v, %v) = %v, want %v", tc.qname, tc.id, tc.score, got, tc.want)
continue continue
} }
gotDead := h.GetDeadMessages(t, r.client) for qname, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortMsgOpt); diff != "" { gotDead := h.GetDeadMessages(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.DeadQueue, diff) if diff := cmp.Diff(want, gotDead, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.DeadKey(qname), diff)
}
} }
} }
} }
func TestDeleteRetryTask(t *testing.T) { func TestDeleteRetryTask(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")
t1 := time.Now().Add(5 * time.Minute) t1 := time.Now().Add(5 * time.Minute)
t2 := time.Now().Add(time.Hour) t2 := time.Now().Add(time.Hour)
t3 := time.Now().Add(time.Hour)
tests := []struct { tests := []struct {
retry []base.Z retry map[string][]base.Z
qname string
id uuid.UUID id uuid.UUID
score int64 score int64
want error want error
wantRetry []*base.TaskMessage wantRetry map[string][]*base.TaskMessage
}{ }{
{ {
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()},
},
},
qname: "default",
id: m1.ID,
score: t1.Unix(),
want: nil,
wantRetry: map[string][]*base.TaskMessage{
"default": {m2},
}, },
id: m1.ID,
score: t1.Unix(),
want: nil,
wantRetry: []*base.TaskMessage{m2},
}, },
{ {
retry: []base.Z{ retry: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
},
},
qname: "custom",
id: m3.ID,
score: t3.Unix(),
want: nil,
wantRetry: map[string][]*base.TaskMessage{
"default": {m1, m2},
"custom": {},
},
},
{
retry: map[string][]base.Z{
"default": {{Message: m1, Score: t1.Unix()}},
},
qname: "default",
id: m2.ID,
score: t2.Unix(),
want: ErrTaskNotFound,
wantRetry: map[string][]*base.TaskMessage{
"default": {m1},
}, },
id: m2.ID,
score: t2.Unix(),
want: ErrTaskNotFound,
wantRetry: []*base.TaskMessage{m1},
}, },
} }
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.SeedRetryQueue(t, r.client, tc.retry) h.SeedAllRetryQueues(t, r.client, tc.retry)
got := r.DeleteRetryTask(tc.id, tc.score) got := r.DeleteRetryTask(tc.qname, tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("r.DeleteRetryTask(%v, %v) = %v, want %v", tc.id, tc.score, got, tc.want) t.Errorf("r.DeleteRetryTask(%q, %v, %v) = %v, want %v", tc.qname, tc.id, tc.score, got, tc.want)
continue continue
} }
gotRetry := h.GetRetryMessages(t, r.client) for qname, want := range tc.wantRetry {
if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortMsgOpt); diff != "" { gotRetry := h.GetRetryMessages(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryQueue, diff) if diff := cmp.Diff(want, gotRetry, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryKey(qname), diff)
}
} }
} }
} }
func TestDeleteScheduledTask(t *testing.T) { func TestDeleteScheduledTask(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")
t1 := time.Now().Add(5 * time.Minute) t1 := time.Now().Add(5 * time.Minute)
t2 := time.Now().Add(time.Hour) t2 := time.Now().Add(time.Hour)
t3 := time.Now().Add(time.Hour)
tests := []struct { tests := []struct {
scheduled []base.Z scheduled map[string][]base.Z
qname string
id uuid.UUID id uuid.UUID
score int64 score int64
want error want error
wantScheduled []*base.TaskMessage wantScheduled map[string][]*base.TaskMessage
}{ }{
{ {
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()},
},
},
qname: "default",
id: m1.ID,
score: t1.Unix(),
want: nil,
wantScheduled: map[string][]*base.TaskMessage{
"default": {m2},
}, },
id: m1.ID,
score: t1.Unix(),
want: nil,
wantScheduled: []*base.TaskMessage{m2},
}, },
{ {
scheduled: []base.Z{ scheduled: map[string][]base.Z{
{Message: m1, Score: t1.Unix()}, "default": {
{Message: m1, Score: t1.Unix()},
{Message: m2, Score: t2.Unix()},
},
"custom": {
{Message: m3, Score: t3.Unix()},
},
},
qname: "custom",
id: m3.ID,
score: t3.Unix(),
want: nil,
wantScheduled: map[string][]*base.TaskMessage{
"default": {m1, m2},
"custom": {},
},
},
{
scheduled: map[string][]base.Z{
"default": {{Message: m1, Score: t1.Unix()}},
},
qname: "default",
id: m2.ID,
score: t2.Unix(),
want: ErrTaskNotFound,
wantScheduled: map[string][]*base.TaskMessage{
"default": {m1},
}, },
id: m2.ID,
score: t2.Unix(),
want: ErrTaskNotFound,
wantScheduled: []*base.TaskMessage{m1},
}, },
} }
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.SeedScheduledQueue(t, r.client, tc.scheduled) h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
got := r.DeleteScheduledTask(tc.id, tc.score) got := r.DeleteScheduledTask(tc.qname, tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("r.DeleteScheduledTask(%v, %v) = %v, want %v", tc.id, tc.score, got, tc.want) t.Errorf("r.DeleteScheduledTask(%q, %v, %v) = %v, want %v", tc.qname, tc.id, tc.score, got, tc.want)
continue continue
} }
gotScheduled := h.GetScheduledMessages(t, r.client) for qname, want := range tc.wantScheduled {
if diff := cmp.Diff(tc.wantScheduled, gotScheduled, h.SortMsgOpt); diff != "" { gotScheduled := h.GetScheduledMessages(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ScheduledQueue, diff) if diff := cmp.Diff(want, gotScheduled, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ScheduledKey(qname), diff)
}
} }
} }
} }
func TestDeleteAllDeadTasks(t *testing.T) { func TestDeleteAllDeadTasks(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.NewTaskMessage("gen_thumbnail", nil) m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
tests := []struct { tests := []struct {
dead []base.Z dead map[string][]base.Z
qname string
want int64 want int64
wantDead []*base.TaskMessage wantDead map[string][]*base.TaskMessage
}{ }{
{ {
dead: []base.Z{ dead: map[string][]base.Z{
{Message: m1, Score: time.Now().Unix()}, "default": {
{Message: m2, Score: time.Now().Unix()}, {Message: m1, Score: time.Now().Unix()},
{Message: m3, Score: time.Now().Unix()}, {Message: m2, Score: time.Now().Unix()},
},
"custom": {
{Message: m3, Score: time.Now().Unix()},
},
},
qname: "default",
want: 2,
wantDead: map[string][]*base.TaskMessage{
"default": {},
"custom": {m3},
}, },
want: 3,
wantDead: []*base.TaskMessage{},
}, },
{ {
dead: []base.Z{}, dead: map[string][]base.Z{
want: 0, "default": {},
wantDead: []*base.TaskMessage{}, },
qname: "default",
want: 0,
wantDead: map[string][]*base.TaskMessage{
"default": {},
},
}, },
} }
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.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
got, err := r.DeleteAllDeadTasks() got, err := r.DeleteAllDeadTasks(tc.qname)
if err != nil { if err != nil {
t.Errorf("r.DeleteAllDeadTasks returned error: %v", err) t.Errorf("r.DeleteAllDeadTasks(%q) returned error: %v", tc.qname, err)
} }
if got != tc.want { if got != tc.want {
t.Errorf("r.DeleteAllDeadTasks() = %d, nil, want %d, nil", got, tc.want) t.Errorf("r.DeleteAllDeadTasks(%q) = %d, nil, want %d, nil", tc.qname, got, tc.want)
} }
gotDead := h.GetDeadMessages(t, r.client) for qname, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortMsgOpt); diff != "" { gotDead := h.GetDeadMessages(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.DeadQueue, diff) if diff := cmp.Diff(want, gotDead, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.DeadKey(qname), diff)
}
} }
} }
} }
func TestDeleteAllRetryTasks(t *testing.T) { func TestDeleteAllRetryTasks(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.NewTaskMessage("gen_thumbnail", nil) m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
tests := []struct { tests := []struct {
retry []base.Z retry map[string][]base.Z
qname string
want int64 want int64
wantRetry []*base.TaskMessage wantRetry map[string][]*base.TaskMessage
}{ }{
{ {
retry: []base.Z{ retry: map[string][]base.Z{
{Message: m1, Score: time.Now().Unix()}, "default": {
{Message: m2, Score: time.Now().Unix()}, {Message: m1, Score: time.Now().Unix()},
{Message: m3, Score: time.Now().Unix()}, {Message: m2, Score: time.Now().Unix()},
},
"custom": {
{Message: m3, Score: time.Now().Unix()},
},
},
qname: "custom",
want: 1,
wantRetry: map[string][]*base.TaskMessage{
"default": {m1, m2},
"custom": {},
}, },
want: 3,
wantRetry: []*base.TaskMessage{},
}, },
{ {
retry: []base.Z{}, retry: map[string][]base.Z{
want: 0, "default": {},
wantRetry: []*base.TaskMessage{}, },
qname: "default",
want: 0,
wantRetry: map[string][]*base.TaskMessage{
"default": {},
},
}, },
} }
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.SeedRetryQueue(t, r.client, tc.retry) h.SeedAllRetryQueues(t, r.client, tc.retry)
got, err := r.DeleteAllRetryTasks() got, err := r.DeleteAllRetryTasks(tc.qname)
if err != nil { if err != nil {
t.Errorf("r.DeleteAllRetryTasks returned error: %v", err) t.Errorf("r.DeleteAllRetryTasks(%q) returned error: %v", tc.qname, err)
} }
if got != tc.want { if got != tc.want {
t.Errorf("r.DeleteAllRetryTasks() = %d, nil, want %d, nil", got, tc.want) t.Errorf("r.DeleteAllRetryTasks(%q) = %d, nil, want %d, nil", tc.qname, got, tc.want)
} }
gotRetry := h.GetRetryMessages(t, r.client) for qname, want := range tc.wantRetry {
if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortMsgOpt); diff != "" { gotRetry := h.GetRetryMessages(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryQueue, diff) if diff := cmp.Diff(want, gotRetry, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryKey(qname), diff)
}
} }
} }
} }
func TestDeleteAllScheduledTasks(t *testing.T) { func TestDeleteAllScheduledTasks(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.NewTaskMessage("gen_thumbnail", nil) m3 := h.NewTaskMessageWithQueue("task3", nil, "custom")
tests := []struct { tests := []struct {
scheduled []base.Z scheduled map[string][]base.Z
qname string
want int64 want int64
wantScheduled []*base.TaskMessage wantScheduled map[string][]*base.TaskMessage
}{ }{
{ {
scheduled: []base.Z{ scheduled: map[string][]base.Z{
{Message: m1, Score: time.Now().Add(time.Minute).Unix()}, "default": {
{Message: m2, Score: time.Now().Add(time.Minute).Unix()}, {Message: m1, Score: time.Now().Add(time.Minute).Unix()},
{Message: m3, Score: time.Now().Add(time.Minute).Unix()}, {Message: m2, Score: time.Now().Add(time.Minute).Unix()},
},
"custom": {
{Message: m3, Score: time.Now().Add(time.Minute).Unix()},
},
},
qname: "default",
want: 2,
wantScheduled: map[string][]*base.TaskMessage{
"default": {},
"custom": {m3},
}, },
want: 3,
wantScheduled: []*base.TaskMessage{},
}, },
{ {
scheduled: []base.Z{}, scheduled: map[string][]base.Z{
want: 0, "custom": {},
wantScheduled: []*base.TaskMessage{}, },
qname: "custom",
want: 0,
wantScheduled: map[string][]*base.TaskMessage{
"custom": {},
},
}, },
} }
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.SeedScheduledQueue(t, r.client, tc.scheduled) h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
got, err := r.DeleteAllScheduledTasks() got, err := r.DeleteAllScheduledTasks(tc.qname)
if err != nil { if err != nil {
t.Errorf("r.DeleteAllScheduledTasks returned error: %v", err) t.Errorf("r.DeleteAllScheduledTasks(%q) returned error: %v", tc.qname, err)
} }
if got != tc.want { if got != tc.want {
t.Errorf("r.DeleteAllScheduledTasks() = %d, nil, want %d, nil", got, tc.want) t.Errorf("r.DeleteAllScheduledTasks(%q) = %d, nil, want %d, nil", tc.qname, got, tc.want)
} }
gotScheduled := h.GetScheduledMessages(t, r.client) for qname, want := range tc.wantScheduled {
if diff := cmp.Diff(tc.wantScheduled, gotScheduled, h.SortMsgOpt); diff != "" { gotScheduled := h.GetScheduledMessages(t, r.client, qname)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ScheduledQueue, diff) if diff := cmp.Diff(want, gotScheduled, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ScheduledKey(qname), diff)
}
} }
} }
} }
/*
func TestRemoveQueue(t *testing.T) { func TestRemoveQueue(t *testing.T) {
r := setup(t) r := setup(t)
m1 := h.NewTaskMessage("send_email", nil) m1 := h.NewTaskMessage("send_email", nil)