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

Rename rdb methods to enqueue scheduled, retry, and dead tasks

Browse Source
This commit is contained in:
Ken Hibino 2019-12-09 19:33:07 -08:00
parent b766de4f18
commit 8830d23388
3 changed files with 21 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
internal/rdb/inspect.go
View File

@ -230,10 +230,10 @@ func (r *RDB) ListDead() ([]*DeadTask, error) {
return tasks, nil return tasks, nil
} }
// Rescue finds a task that matches the given id and score from dead queue // EnqueueDeadTask finds a task that matches the given id and score from dead queue
// and enqueues it for processing. If a task that matches the id and score // and enqueues it for processing. If a task that matches the id and score
// does not exist, it returns ErrTaskNotFound. // does not exist, it returns ErrTaskNotFound.
func (r *RDB) Rescue(id string, score float64) error { func (r *RDB) EnqueueDeadTask(id string, score float64) error {
n, err := r.removeAndEnqueue(deadQ, id, score) n, err := r.removeAndEnqueue(deadQ, id, score)
if err != nil { if err != nil {
return err return err
@ -244,10 +244,10 @@ func (r *RDB) Rescue(id string, score float64) error {
return nil return nil
} }
// RetryNow finds a task that matches the given id and score from retry queue // EnqueueRetryTask finds a task that matches the given id and score from retry queue
// and enqueues it for processing. If a task that matches the id and score // and enqueues it for processing. If a task that matches the id and score
// does not exist, it returns ErrTaskNotFound. // does not exist, it returns ErrTaskNotFound.
func (r *RDB) RetryNow(id string, score float64) error { func (r *RDB) EnqueueRetryTask(id string, score float64) error {
n, err := r.removeAndEnqueue(retryQ, id, score) n, err := r.removeAndEnqueue(retryQ, id, score)
if err != nil { if err != nil {
return err return err
@ -258,10 +258,10 @@ func (r *RDB) RetryNow(id string, score float64) error {
return nil return nil
} }
// ProcessNow finds a task that matches the given id and score from scheduled queue // EnqueueScheduledTask finds a task that matches the given id and score from scheduled queue
// and enqueues it for processing. If a task that matches the id and score does not // and enqueues it for processing. If a task that matches the id and score does not
// exist, it returns ErrTaskNotFound. // exist, it returns ErrTaskNotFound.
func (r *RDB) ProcessNow(id string, score float64) error { func (r *RDB) EnqueueScheduledTask(id string, score float64) error {
n, err := r.removeAndEnqueue(scheduledQ, id, score) n, err := r.removeAndEnqueue(scheduledQ, id, score)
if err != nil { if err != nil {
return err return err

24
internal/rdb/inspect_test.go
View File

@ -471,7 +471,7 @@ func TestListDead(t *testing.T) {
var timeCmpOpt = EquateApproxTime(time.Second) var timeCmpOpt = EquateApproxTime(time.Second)
func TestRescue(t *testing.T) { func TestEnqueueDeadTask(t *testing.T) {
r := setup(t) r := setup(t)
t1 := randomTask("send_email", "default", nil) t1 := randomTask("send_email", "default", nil)
@ -486,7 +486,7 @@ func TestRescue(t *testing.T) {
dead []deadEntry dead []deadEntry
score float64 score float64
id string id string
want error // expected return value from calling Rescue want error // expected return value from calling EnqueueDeadTask
wantDead []*TaskMessage wantDead []*TaskMessage
wantEnqueued []*TaskMessage wantEnqueued []*TaskMessage
}{ }{
@ -527,9 +527,9 @@ func TestRescue(t *testing.T) {
} }
} }
got := r.Rescue(tc.id, tc.score) got := r.EnqueueDeadTask(tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("r.Rescue(%s, %0.f) = %v, want %v", tc.id, tc.score, got, tc.want) t.Errorf("r.EnqueueDeadTask(%s, %0.f) = %v, want %v", tc.id, tc.score, got, tc.want)
continue continue
} }
@ -547,7 +547,7 @@ func TestRescue(t *testing.T) {
} }
} }
func TestRetryNow(t *testing.T) { func TestEnqueueRetryTask(t *testing.T) {
r := setup(t) r := setup(t)
t1 := randomTask("send_email", "default", nil) t1 := randomTask("send_email", "default", nil)
@ -562,7 +562,7 @@ func TestRetryNow(t *testing.T) {
dead []retryEntry dead []retryEntry
score float64 score float64
id string id string
want error // expected return value from calling RetryNow want error // expected return value from calling EnqueueRetryTask
wantRetry []*TaskMessage wantRetry []*TaskMessage
wantEnqueued []*TaskMessage wantEnqueued []*TaskMessage
}{ }{
@ -603,9 +603,9 @@ func TestRetryNow(t *testing.T) {
} }
} }
got := r.RetryNow(tc.id, tc.score) got := r.EnqueueRetryTask(tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("r.RetryNow(%s, %0.f) = %v, want %v", tc.id, tc.score, got, tc.want) t.Errorf("r.EnqueueRetryTask(%s, %0.f) = %v, want %v", tc.id, tc.score, got, tc.want)
continue continue
} }
@ -623,7 +623,7 @@ func TestRetryNow(t *testing.T) {
} }
} }
func TestProcessNow(t *testing.T) { func TestEnqueueScheduledTask(t *testing.T) {
r := setup(t) r := setup(t)
t1 := randomTask("send_email", "default", nil) t1 := randomTask("send_email", "default", nil)
@ -638,7 +638,7 @@ func TestProcessNow(t *testing.T) {
dead []scheduledEntry dead []scheduledEntry
score float64 score float64
id string id string
want error // expected return value from calling ProcessNow want error // expected return value from calling EnqueueScheduledTask
wantScheduled []*TaskMessage wantScheduled []*TaskMessage
wantEnqueued []*TaskMessage wantEnqueued []*TaskMessage
}{ }{
@ -679,9 +679,9 @@ func TestProcessNow(t *testing.T) {
} }
} }
got := r.ProcessNow(tc.id, tc.score) got := r.EnqueueScheduledTask(tc.id, tc.score)
if got != tc.want { if got != tc.want {
t.Errorf("r.RetryNow(%s, %0.f) = %v, want %v", tc.id, tc.score, got, tc.want) t.Errorf("r.EnqueueRetryTask(%s, %0.f) = %v, want %v", tc.id, tc.score, got, tc.want)
continue continue
} }

6
tools/asynqmon/cmd/enq.go
View File

@ -53,11 +53,11 @@ func enq(cmd *cobra.Command, args []string) {
})) }))
switch qtype { switch qtype {
case "s": case "s":
err = r.ProcessNow(id.String(), float64(score)) err = r.EnqueueScheduledTask(id.String(), float64(score))
case "r": case "r":
err = r.RetryNow(id.String(), float64(score)) err = r.EnqueueRetryTask(id.String(), float64(score))
case "d": case "d":
err = r.Rescue(id.String(), float64(score)) err = r.EnqueueDeadTask(id.String(), float64(score))
default: default:
fmt.Println("invalid argument") fmt.Println("invalid argument")
os.Exit(1) os.Exit(1)