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

Remove Queue field and Priority field to base.TaskMessage

Browse Source
This commit is contained in:
Ken Hibino 2020-01-01 07:48:21 -08:00
parent 31a18d783e
commit 2e95166f39
5 changed files with 33 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
client.go
View File

@ -76,11 +76,11 @@ const (
func (c *Client) Process(task *Task, processAt time.Time, opts ...Option) error { func (c *Client) Process(task *Task, processAt time.Time, opts ...Option) error {
opt := composeOptions(opts...) opt := composeOptions(opts...)
msg := &base.TaskMessage{ msg := &base.TaskMessage{
ID: xid.New(), ID: xid.New(),
Type: task.Type, Type: task.Type,
Payload: task.Payload, Payload: task.Payload,
Queue: "default", Priority: base.PriorityDefault,
Retry: opt.retry, Retry: opt.retry,
} }
return c.enqueue(msg, processAt) return c.enqueue(msg, processAt)
} }

40
client_test.go
View File

@ -30,10 +30,10 @@ func TestClient(t *testing.T) {
opts: []Option{}, opts: []Option{},
wantEnqueued: []*base.TaskMessage{ wantEnqueued: []*base.TaskMessage{
&base.TaskMessage{ &base.TaskMessage{
Type: task.Type, Type: task.Type,
Payload: task.Payload, Payload: task.Payload,
Retry: defaultMaxRetry, Retry: defaultMaxRetry,
Queue: "default", Priority: base.PriorityDefault,
}, },
}, },
wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil
@ -47,10 +47,10 @@ func TestClient(t *testing.T) {
wantScheduled: []h.ZSetEntry{ wantScheduled: []h.ZSetEntry{
{ {
Msg: &base.TaskMessage{ Msg: &base.TaskMessage{
Type: task.Type, Type: task.Type,
Payload: task.Payload, Payload: task.Payload,
Retry: defaultMaxRetry, Retry: defaultMaxRetry,
Queue: "default", Priority: base.PriorityDefault,
}, },
Score: time.Now().Add(2 * time.Hour).Unix(), Score: time.Now().Add(2 * time.Hour).Unix(),
}, },
@ -65,10 +65,10 @@ func TestClient(t *testing.T) {
}, },
wantEnqueued: []*base.TaskMessage{ wantEnqueued: []*base.TaskMessage{
&base.TaskMessage{ &base.TaskMessage{
Type: task.Type, Type: task.Type,
Payload: task.Payload, Payload: task.Payload,
Retry: 3, Retry: 3,
Queue: "default", Priority: base.PriorityDefault,
}, },
}, },
wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil
@ -82,10 +82,10 @@ func TestClient(t *testing.T) {
}, },
wantEnqueued: []*base.TaskMessage{ wantEnqueued: []*base.TaskMessage{
&base.TaskMessage{ &base.TaskMessage{
Type: task.Type, Type: task.Type,
Payload: task.Payload, Payload: task.Payload,
Retry: 0, // Retry count should be set to zero Retry: 0, // Retry count should be set to zero
Queue: "default", Priority: base.PriorityDefault,
}, },
}, },
wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil
@ -100,10 +100,10 @@ func TestClient(t *testing.T) {
}, },
wantEnqueued: []*base.TaskMessage{ wantEnqueued: []*base.TaskMessage{
&base.TaskMessage{ &base.TaskMessage{
Type: task.Type, Type: task.Type,
Payload: task.Payload, Payload: task.Payload,
Retry: 10, // Last option takes precedence Retry: 10, // Last option takes precedence
Queue: "default", Priority: base.PriorityDefault,
}, },
}, },
wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil

10
internal/asynqtest/asynqtest.go
View File

@ -43,11 +43,11 @@ var IgnoreIDOpt = cmpopts.IgnoreFields(base.TaskMessage{}, "ID")
// NewTaskMessage returns a new instance of TaskMessage given a task type and payload. // NewTaskMessage returns a new instance of TaskMessage given a task type and payload.
func NewTaskMessage(taskType string, payload map[string]interface{}) *base.TaskMessage { func NewTaskMessage(taskType string, payload map[string]interface{}) *base.TaskMessage {
return &base.TaskMessage{ return &base.TaskMessage{
ID: xid.New(), ID: xid.New(),
Type: taskType, Type: taskType,
Queue: "default", Priority: base.PriorityDefault,
Retry: 25, Retry: 25,
Payload: payload, Payload: payload,
} }
} }

4
internal/base/base.go
View File

@ -62,8 +62,8 @@ type TaskMessage struct {
// ID is a unique identifier for each task. // ID is a unique identifier for each task.
ID xid.ID ID xid.ID
// Queue is a name this message should be enqueued to. // Priority is the priority of this task.
Queue string Priority Priority
// Retry is the max number of retry for this task. // Retry is the max number of retry for this task.
Retry int Retry int

2
internal/rdb/rdb.go
View File

@ -42,7 +42,7 @@ func (r *RDB) Enqueue(msg *base.TaskMessage) error {
if err != nil { if err != nil {
return err return err
} }
qname := base.QueuePrefix + msg.Queue qname := base.QueueKey(msg.Priority)
return r.client.LPush(qname, string(bytes)).Err() return r.client.LPush(qname, string(bytes)).Err()
} }