go-pkg/asynq
1
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2025-09-17 20:30:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Run task by qname and id

Browse Source
This commit is contained in:
Ken Hibino
2021-03-28 06:58:54 -07:00
parent 4bb67c582d
commit 56af9f6686
4 changed files with 97 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
internal/rdb/inspect.go
View File

@@ -10,7 +10,6 @@ import (
"time"
"github.com/go-redis/redis/v7"
"github.com/google/uuid"
"github.com/hibiken/asynq/internal/base"
"github.com/spf13/cast"
)
@@ -447,48 +446,6 @@ func (r *RDB) listZSetEntries(key, qname string, pgn Pagination) ([]base.Z, erro
return zs, nil
}
// RunArchivedTask finds an archived task that matches the given id and score from
// the given queue and enqueues it for processing.
// If a task that matches the id and score does not exist, it returns ErrTaskNotFound.
func (r *RDB) RunArchivedTask(qname string, id uuid.UUID) error {
n, err := r.removeAndRun(base.ArchivedKey(qname), base.PendingKey(qname), id.String())
if err != nil {
return err
}
if n == 0 {
return ErrTaskNotFound
}
return nil
}
// RunRetryTask finds a retry task that matches the given id and score from
// the given queue and enqueues it for processing.
// If a task that matches the id and score does not exist, it returns ErrTaskNotFound.
func (r *RDB) RunRetryTask(qname string, id uuid.UUID) error {
n, err := r.removeAndRun(base.RetryKey(qname), base.PendingKey(qname), id.String())
if err != nil {
return err
}
if n == 0 {
return ErrTaskNotFound
}
return nil
}
// RunScheduledTask finds a scheduled task that matches the given id and score from
// from the given queue and enqueues it for processing.
// If a task that matches the id and score does not exist, it returns ErrTaskNotFound.
func (r *RDB) RunScheduledTask(qname string, id uuid.UUID) error {
n, err := r.removeAndRun(base.ScheduledKey(qname), base.PendingKey(qname), id.String())
if err != nil {
return err
}
if n == 0 {
return ErrTaskNotFound
}
return nil
}
// RunAllScheduledTasks enqueues all scheduled tasks from the given queue
// and returns the number of tasks enqueued.
func (r *RDB) RunAllScheduledTasks(qname string) (int64, error) {
@@ -519,16 +476,60 @@ redis.call("LPUSH", KEYS[2], ARGV[1])
return 1
`)
func (r *RDB) removeAndRun(zset, qkey, id string) (int64, error) {
res, err := removeAndRunCmd.Run(r.client, []string{zset, qkey}, id).Result()
// KEYS[1] -> asynq:{<qname>}:t:<task_id>
// KEYS[2] -> asynq:{<qname>}:pending
// ARGV[1] -> task ID
// ARGV[2] -> redis key prefix (asynq:{<qname>}:)
var runTaskCmd = redis.NewScript(`
if redis.call("EXISTS", KEYS[1]) == 0 then
return 0
end
local state = redis.call("HGET", KEYS[1], "state")
local n = 0
if state == "ACTIVE" then
return redis.error_reply("task is already active")
elseif state == "PENDING" then
return redis.error_reply("task is already pending")
elseif state == "SCHEDULED" then
n = redis.call("ZREM", (ARGV[2] .. "scheduled"), ARGV[1])
elseif state == "RETRY" then
n = redis.call("ZREM", (ARGV[2] .. "retry"), ARGV[1])
elseif state == "ARCHIVED" then
n = redis.call("ZREM", (ARGV[2] .. "archived"), ARGV[1])
else
return redis.error_reply("unknown task state: " .. tostring(state))
end
if n == 0 then
return 0
end
redis.call("LPUSH", KEYS[2], ARGV[1])
return 1
`)
// RunTask finds a task that matches the given id from the given queue
// and stage it for processing (i.e. transition the task to pending state).
// If no match is found, it returns ErrTaskNotFound.
func (r *RDB) RunTask(qname, id string) error {
keys := []string{
base.TaskKey(qname, id),
base.PendingKey(qname),
}
argv := []interface{}{
id,
base.QueueKeyPrefix(qname),
}
res, err := runTaskCmd.Run(r.client, keys, argv...).Result()
if err != nil {
return 0, err
return err
}
n, ok := res.(int64)
if !ok {
return 0, fmt.Errorf("could not cast %v to int64", res)
return fmt.Errorf("command error: unexpected return value %v", res)
}
return n, nil
if n == 0 {
return ErrTaskNotFound
}
return nil
}
var removeAndRunAllCmd = redis.NewScript(`
@@ -712,7 +713,7 @@ elseif state == "RETRY" then
n = redis.call("ZREM", (ARGV[2] .. "retry"), ARGV[1])
elseif state == "ARCHIVED" then
n = redis.call("ZREM", (ARGV[2] .. "archived"), ARGV[1])
elseif state == "ACTIVE"
elseif state == "ACTIVE" then
return redis.error_reply("cannot delete active task")
else
return redis.error_reply("unknown task state: " .. tostring(state))

12
internal/rdb/inspect_test.go
View File

@@ -1075,9 +1075,9 @@ func TestRunArchivedTask(t *testing.T) {
h.FlushDB(t, r.client) // clean up db before each test case
h.SeedAllArchivedQueues(t, r.client, tc.archived)
got := r.RunArchivedTask(tc.qname, tc.id)
got := r.RunTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.RunDeadTask(%q, %s) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.RunTask(%q, %s) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}
@@ -1176,9 +1176,9 @@ func TestRunRetryTask(t *testing.T) {
h.FlushDB(t, r.client) // clean up db before each test case
h.SeedAllRetryQueues(t, r.client, tc.retry) // initialize retry queue
got := r.RunRetryTask(tc.qname, tc.id)
got := r.RunTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.RunRetryTask(%q, %s) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.RunTask(%q, %s) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}
@@ -1277,9 +1277,9 @@ func TestRunScheduledTask(t *testing.T) {
h.FlushDB(t, r.client) // clean up db before each test case
h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
got := r.RunScheduledTask(tc.qname, tc.id)
got := r.RunTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.RunRetryTask(%q, %s) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.RunTask(%q, %s) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}