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

Replace github.com/rs/xid with github.com/google/uuid

Browse Source
This commit is contained in:
Ken Hibino
2020-07-02 06:21:20 -07:00
parent 486dcd799b
commit 8b60e6a268
15 changed files with 104 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
internal/asynqtest/asynqtest.go
View File

@@ -13,8 +13,8 @@ import (
"github.com/go-redis/redis/v7"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
"github.com/hibiken/asynq/internal/base"
"github.com/rs/xid"
)
// ZSetEntry is an entry in redis sorted set.
@@ -75,7 +75,7 @@ var IgnoreIDOpt = cmpopts.IgnoreFields(base.TaskMessage{}, "ID")
// NewTaskMessage returns a new instance of TaskMessage given a task type and payload.
func NewTaskMessage(taskType string, payload map[string]interface{}) *base.TaskMessage {
return &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: taskType,
Queue: base.DefaultQueueName,
Retry: 25,
@@ -89,7 +89,7 @@ func NewTaskMessage(taskType string, payload map[string]interface{}) *base.TaskM
// task type, payload and queue name.
func NewTaskMessageWithQueue(taskType string, payload map[string]interface{}, qname string) *base.TaskMessage {
return &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: taskType,
Queue: qname,
Retry: 25,

4
internal/base/base.go
View File

@@ -14,7 +14,7 @@ import (
"time"
"github.com/go-redis/redis/v7"
"github.com/rs/xid"
"github.com/google/uuid"
)
// Version of asynq library and CLI.
@@ -78,7 +78,7 @@ type TaskMessage struct {
Payload map[string]interface{}
// ID is a unique identifier for each task.
ID xid.ID
ID uuid.UUID
// Queue is a name this message should be enqueued to.
Queue string

4
internal/base/base_test.go
View File

@@ -12,7 +12,7 @@ import (
"time"
"github.com/google/go-cmp/cmp"
"github.com/rs/xid"
"github.com/google/uuid"
)
func TestQueueKey(t *testing.T) {
@@ -108,7 +108,7 @@ func TestWorkersKey(t *testing.T) {
}
func TestMessageEncoding(t *testing.T) {
id := xid.New()
id := uuid.New()
tests := []struct {
in *TaskMessage
out *TaskMessage

28
internal/rdb/inspect.go
View File

@@ -12,8 +12,8 @@ import (
"time"
"github.com/go-redis/redis/v7"
"github.com/google/uuid"
"github.com/hibiken/asynq/internal/base"
"github.com/rs/xid"
"github.com/spf13/cast"
)
@@ -53,7 +53,7 @@ type DailyStats struct {
// EnqueuedTask is a task in a queue and is ready to be processed.
type EnqueuedTask struct {
ID xid.ID
ID uuid.UUID
Type string
Payload map[string]interface{}
Queue string
@@ -61,14 +61,14 @@ type EnqueuedTask struct {
// InProgressTask is a task that's currently being processed.
type InProgressTask struct {
ID xid.ID
ID uuid.UUID
Type string
Payload map[string]interface{}
}
// ScheduledTask is a task that's scheduled to be processed in the future.
type ScheduledTask struct {
ID xid.ID
ID uuid.UUID
Type string
Payload map[string]interface{}
ProcessAt time.Time
@@ -78,7 +78,7 @@ type ScheduledTask struct {
// RetryTask is a task that's in retry queue because worker failed to process the task.
type RetryTask struct {
ID xid.ID
ID uuid.UUID
Type string
Payload map[string]interface{}
// TODO(hibiken): add LastFailedAt time.Time
@@ -92,7 +92,7 @@ type RetryTask struct {
// DeadTask is a task in that has exhausted all retries.
type DeadTask struct {
ID xid.ID
ID uuid.UUID
Type string
Payload map[string]interface{}
LastFailedAt time.Time
@@ -446,7 +446,7 @@ func (r *RDB) ListDead(pgn Pagination) ([]*DeadTask, error) {
// 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
// does not exist, it returns ErrTaskNotFound.
func (r *RDB) EnqueueDeadTask(id xid.ID, score int64) error {
func (r *RDB) EnqueueDeadTask(id uuid.UUID, score int64) error {
n, err := r.removeAndEnqueue(base.DeadQueue, id.String(), float64(score))
if err != nil {
return err
@@ -460,7 +460,7 @@ func (r *RDB) EnqueueDeadTask(id xid.ID, score int64) error {
// 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
// does not exist, it returns ErrTaskNotFound.
func (r *RDB) EnqueueRetryTask(id xid.ID, score int64) error {
func (r *RDB) EnqueueRetryTask(id uuid.UUID, score int64) error {
n, err := r.removeAndEnqueue(base.RetryQueue, id.String(), float64(score))
if err != nil {
return err
@@ -474,7 +474,7 @@ func (r *RDB) EnqueueRetryTask(id xid.ID, score int64) error {
// 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
// exist, it returns ErrTaskNotFound.
func (r *RDB) EnqueueScheduledTask(id xid.ID, score int64) error {
func (r *RDB) EnqueueScheduledTask(id uuid.UUID, score int64) error {
n, err := r.removeAndEnqueue(base.ScheduledQueue, id.String(), float64(score))
if err != nil {
return err
@@ -553,7 +553,7 @@ func (r *RDB) removeAndEnqueueAll(zset string) (int64, error) {
// KillRetryTask finds a task that matches the given id and score from retry queue
// and moves it to dead queue. If a task that maches the id and score does not exist,
// it returns ErrTaskNotFound.
func (r *RDB) KillRetryTask(id xid.ID, score int64) error {
func (r *RDB) KillRetryTask(id uuid.UUID, score int64) error {
n, err := r.removeAndKill(base.RetryQueue, id.String(), float64(score))
if err != nil {
return err
@@ -567,7 +567,7 @@ func (r *RDB) KillRetryTask(id xid.ID, score int64) error {
// KillScheduledTask finds a task that matches the given id and score from scheduled queue
// and moves it to dead queue. If a task that maches the id and score does not exist,
// it returns ErrTaskNotFound.
func (r *RDB) KillScheduledTask(id xid.ID, score int64) error {
func (r *RDB) KillScheduledTask(id uuid.UUID, score int64) error {
n, err := r.removeAndKill(base.ScheduledQueue, id.String(), float64(score))
if err != nil {
return err
@@ -660,21 +660,21 @@ func (r *RDB) removeAndKillAll(zset string) (int64, error) {
// 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,
// it returns ErrTaskNotFound.
func (r *RDB) DeleteDeadTask(id xid.ID, score int64) error {
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
// and deletes it. If a task that matches the id and score does not exist,
// it returns ErrTaskNotFound.
func (r *RDB) DeleteRetryTask(id xid.ID, score int64) error {
func (r *RDB) DeleteRetryTask(id uuid.UUID, score int64) error {
return r.deleteTask(base.RetryQueue, id.String(), float64(score))
}
// DeleteScheduledTask finds a task that matches the given id and score from
// scheduled queue and deletes it. If a task that matches the id and score
//does not exist, it returns ErrTaskNotFound.
func (r *RDB) DeleteScheduledTask(id xid.ID, score int64) error {
func (r *RDB) DeleteScheduledTask(id uuid.UUID, score int64) error {
return r.deleteTask(base.ScheduledQueue, id.String(), float64(score))
}

26
internal/rdb/inspect_test.go
View File

@@ -12,9 +12,9 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
h "github.com/hibiken/asynq/internal/asynqtest"
"github.com/hibiken/asynq/internal/base"
"github.com/rs/xid"
)
func TestCurrentStats(t *testing.T) {
@@ -618,7 +618,7 @@ func TestListScheduledPagination(t *testing.T) {
func TestListRetry(t *testing.T) {
r := setup(t)
m1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Queue: "default",
Payload: map[string]interface{}{"subject": "hello"},
@@ -627,7 +627,7 @@ func TestListRetry(t *testing.T) {
Retried: 10,
}
m2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "reindex",
Queue: "default",
Payload: nil,
@@ -763,14 +763,14 @@ func TestListRetryPagination(t *testing.T) {
func TestListDead(t *testing.T) {
r := setup(t)
m1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Queue: "default",
Payload: map[string]interface{}{"subject": "hello"},
ErrorMsg: "email server not responding",
}
m2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "reindex",
Queue: "default",
Payload: nil,
@@ -907,7 +907,7 @@ func TestEnqueueDeadTask(t *testing.T) {
tests := []struct {
dead []h.ZSetEntry
score int64
id xid.ID
id uuid.UUID
want error // expected return value from calling EnqueueDeadTask
wantDead []*base.TaskMessage
wantEnqueued map[string][]*base.TaskMessage
@@ -991,7 +991,7 @@ func TestEnqueueRetryTask(t *testing.T) {
tests := []struct {
retry []h.ZSetEntry
score int64
id xid.ID
id uuid.UUID
want error // expected return value from calling EnqueueRetryTask
wantRetry []*base.TaskMessage
wantEnqueued map[string][]*base.TaskMessage
@@ -1075,7 +1075,7 @@ func TestEnqueueScheduledTask(t *testing.T) {
tests := []struct {
scheduled []h.ZSetEntry
score int64
id xid.ID
id uuid.UUID
want error // expected return value from calling EnqueueScheduledTask
wantScheduled []*base.TaskMessage
wantEnqueued map[string][]*base.TaskMessage
@@ -1394,7 +1394,7 @@ func TestKillRetryTask(t *testing.T) {
tests := []struct {
retry []h.ZSetEntry
dead []h.ZSetEntry
id xid.ID
id uuid.UUID
score int64
want error
wantRetry []h.ZSetEntry
@@ -1471,7 +1471,7 @@ func TestKillScheduledTask(t *testing.T) {
tests := []struct {
scheduled []h.ZSetEntry
dead []h.ZSetEntry
id xid.ID
id uuid.UUID
score int64
want error
wantScheduled []h.ZSetEntry
@@ -1711,7 +1711,7 @@ func TestDeleteDeadTask(t *testing.T) {
tests := []struct {
dead []h.ZSetEntry
id xid.ID
id uuid.UUID
score int64
want error
wantDead []*base.TaskMessage
@@ -1771,7 +1771,7 @@ func TestDeleteRetryTask(t *testing.T) {
tests := []struct {
retry []h.ZSetEntry
id xid.ID
id uuid.UUID
score int64
want error
wantRetry []*base.TaskMessage
@@ -1823,7 +1823,7 @@ func TestDeleteScheduledTask(t *testing.T) {
tests := []struct {
scheduled []h.ZSetEntry
id xid.ID
id uuid.UUID
score int64
want error
wantScheduled []*base.TaskMessage

36
internal/rdb/rdb_test.go
View File

@@ -14,9 +14,9 @@ import (
"github.com/go-redis/redis/v7"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
h "github.com/hibiken/asynq/internal/asynqtest"
"github.com/hibiken/asynq/internal/base"
"github.com/rs/xid"
)
// TODO(hibiken): Get Redis address and db number from ENV variables.
@@ -73,7 +73,7 @@ func TestEnqueue(t *testing.T) {
func TestEnqueueUnique(t *testing.T) {
r := setup(t)
m1 := base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "email",
Payload: map[string]interface{}{"user_id": 123},
Queue: base.DefaultQueueName,
@@ -116,7 +116,7 @@ func TestDequeue(t *testing.T) {
r := setup(t)
now := time.Now()
t1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Payload: map[string]interface{}{"subject": "hello!"},
Timeout: 1800,
@@ -124,7 +124,7 @@ func TestDequeue(t *testing.T) {
}
t1Deadline := now.Unix() + t1.Timeout
t2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "export_csv",
Payload: nil,
Timeout: 0,
@@ -132,7 +132,7 @@ func TestDequeue(t *testing.T) {
}
t2Deadline := t2.Deadline
t3 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "reindex",
Payload: nil,
Timeout: int64((5 * time.Minute).Seconds()),
@@ -384,7 +384,7 @@ func TestDone(t *testing.T) {
r := setup(t)
now := time.Now()
t1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Payload: nil,
Timeout: 1800,
@@ -392,7 +392,7 @@ func TestDone(t *testing.T) {
}
t1Deadline := now.Unix() + t1.Timeout
t2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "export_csv",
Payload: nil,
Timeout: 0,
@@ -400,7 +400,7 @@ func TestDone(t *testing.T) {
}
t2Deadline := t2.Deadline
t3 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "reindex",
Payload: nil,
Timeout: 1800,
@@ -533,21 +533,21 @@ func TestRequeue(t *testing.T) {
r := setup(t)
now := time.Now()
t1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Payload: nil,
Queue: "default",
Timeout: 1800,
}
t2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "export_csv",
Payload: nil,
Queue: "default",
Timeout: 3000,
}
t3 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Payload: nil,
Queue: "critical",
@@ -688,7 +688,7 @@ func TestSchedule(t *testing.T) {
func TestScheduleUnique(t *testing.T) {
r := setup(t)
m1 := base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "email",
Payload: map[string]interface{}{"user_id": 123},
Queue: base.DefaultQueueName,
@@ -741,7 +741,7 @@ func TestRetry(t *testing.T) {
r := setup(t)
now := time.Now()
t1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Payload: map[string]interface{}{"subject": "Hola!"},
Retried: 10,
@@ -749,14 +749,14 @@ func TestRetry(t *testing.T) {
}
t1Deadline := now.Unix() + t1.Timeout
t2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "gen_thumbnail",
Payload: map[string]interface{}{"path": "some/path/to/image.jpg"},
Timeout: 3000,
}
t2Deadline := now.Unix() + t2.Timeout
t3 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "reindex",
Payload: nil,
Timeout: 60,
@@ -857,7 +857,7 @@ func TestKill(t *testing.T) {
r := setup(t)
now := time.Now()
t1 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "send_email",
Payload: nil,
Queue: "default",
@@ -867,7 +867,7 @@ func TestKill(t *testing.T) {
}
t1Deadline := now.Unix() + t1.Timeout
t2 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "reindex",
Payload: nil,
Queue: "default",
@@ -877,7 +877,7 @@ func TestKill(t *testing.T) {
}
t2Deadline := now.Unix() + t2.Timeout
t3 := &base.TaskMessage{
ID: xid.New(),
ID: uuid.New(),
Type: "generate_csv",
Payload: nil,
Queue: "default",