2
0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-27 00:02:19 +08:00

Return error if queue name is empty when enqueueing

This commit is contained in:
Ken Hibino 2020-08-31 05:39:45 -07:00
parent 4897dba397
commit 131ac823fd
2 changed files with 45 additions and 5 deletions

View File

@ -115,7 +115,11 @@ type option struct {
uniqueTTL time.Duration uniqueTTL time.Duration
} }
func composeOptions(opts ...Option) option { // composeOptions merges user provided options into the default options
// and returns the composed option.
// It also validates the user provided options and returns an error if any of
// the user provided options fail the validations.
func composeOptions(opts ...Option) (option, error) {
res := option{ res := option{
retry: defaultMaxRetry, retry: defaultMaxRetry,
queue: base.DefaultQueueName, queue: base.DefaultQueueName,
@ -127,7 +131,11 @@ func composeOptions(opts ...Option) option {
case retryOption: case retryOption:
res.retry = int(opt) res.retry = int(opt)
case queueOption: case queueOption:
res.queue = string(opt) trimmed := strings.TrimSpace(string(opt))
if len(trimmed) == 0 {
return option{}, fmt.Errorf("queue name must contain one or more characters")
}
res.queue = trimmed
case timeoutOption: case timeoutOption:
res.timeout = time.Duration(opt) res.timeout = time.Duration(opt)
case deadlineOption: case deadlineOption:
@ -138,7 +146,7 @@ func composeOptions(opts ...Option) option {
// ignore unexpected option // ignore unexpected option
} }
} }
return res return res, nil
} }
const ( const (
@ -239,7 +247,10 @@ func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) (*Result, er
if defaults, ok := c.opts[task.Type]; ok { if defaults, ok := c.opts[task.Type]; ok {
opts = append(defaults, opts...) opts = append(defaults, opts...)
} }
opt := composeOptions(opts...) opt, err := composeOptions(opts...)
if err != nil {
return nil, err
}
deadline := noDeadline deadline := noDeadline
if !opt.deadline.IsZero() { if !opt.deadline.IsZero() {
deadline = opt.deadline deadline = opt.deadline
@ -266,7 +277,6 @@ func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) (*Result, er
Timeout: int64(timeout.Seconds()), Timeout: int64(timeout.Seconds()),
UniqueKey: uniqueKey, UniqueKey: uniqueKey,
} }
var err error
now := time.Now() now := time.Now()
if t.Before(now) || t.Equal(now) { if t.Before(now) || t.Equal(now) {
err = c.enqueue(msg, opt.uniqueTTL) err = c.enqueue(msg, opt.uniqueTTL)

View File

@ -463,6 +463,36 @@ func TestClientEnqueueIn(t *testing.T) {
} }
} }
func TestClientEnqueueError(t *testing.T) {
r := setup(t)
client := NewClient(getRedisConnOpt(t))
task := NewTask("send_email", map[string]interface{}{"to": "customer@gmail.com", "from": "merchant@example.com"})
tests := []struct {
desc string
task *Task
opts []Option
}{
{
desc: "With empty queue name",
task: task,
opts: []Option{
Queue(""),
},
},
}
for _, tc := range tests {
h.FlushDB(t, r)
_, err := client.Enqueue(tc.task, tc.opts...)
if err == nil {
t.Errorf("%s; client.Enqueue(task, opts...) did not return non-nil error", tc.desc)
}
}
}
func TestClientDefaultOptions(t *testing.T) { func TestClientDefaultOptions(t *testing.T) {
r := setup(t) r := setup(t)