Rework API

Author: vmihailenco

.golangci.yml

@@ -0,0 +1,2 @@
+run:
+  tests: false

azsqs/factory.go

@@ -16,13 +16,13 @@ type factory struct {
 
 var _ taskq.Factory = (*factory)(nil)
 
-func (f *factory) NewQueue(opt *taskq.QueueOptions) taskq.Queue {
+func (f *factory) NewQueue(opt *taskq.QueueOptions) taskq.Queuer {
 	q := NewQueue(f.sqs, f.accountID, opt)
 	f.base.Add(q)
 	return q
 }
 
-func (f *factory) Queues() []taskq.Queue {
+func (f *factory) Queues() []taskq.Queuer {
 	return f.base.Queues()
 }
 

azsqs/queue.go

@@ -24,8 +24,7 @@ const msgSizeLimit = 262144
 const delayUntilAttr = "TaskqDelayUntil"
 
 type Queue struct {
-	base base.Queue
-	opt  *taskq.QueueOptions
+	opt *taskq.QueueOptions
 
 	sqs       *sqs.SQS
 	accountID string
@@ -44,7 +43,7 @@ type Queue struct {
 	consumer *taskq.Consumer
 }
 
-var _ taskq.Queue = (*Queue)(nil)
+var _ taskq.Queuer = (*Queue)(nil)
 
 func NewQueue(sqs *sqs.SQS, accountID string, opt *taskq.QueueOptions) *Queue {
 	opt.Init()
@@ -62,15 +61,16 @@ func NewQueue(sqs *sqs.SQS, accountID string, opt *taskq.QueueOptions) *Queue {
 }
 
 func (q *Queue) initAddQueue() {
+	queueName := "azsqs:" + q.opt.Name + ":add"
 	q.addQueue = memqueue.NewQueue(&taskq.QueueOptions{
-		Name:       "azsqs:" + q.opt.Name + ":add",
+		Name:       queueName,
 		BufferSize: 100,
 		Redis:      q.opt.Redis,
 	})
-	q.addTask = q.addQueue.NewTask(&taskq.TaskOptions{
-		Name:            "add-message",
+	q.addTask = taskq.NewTask(&taskq.TaskOptions{
+		Name:            queueName + ":add-message",
 		Handler:         taskq.HandlerFunc(q.addBatcherAdd),
-		FallbackHandler: msgutil.UnwrapMessageHandler(q.HandleMessage),
+		FallbackHandler: msgutil.UnwrapMessageHandler(taskq.Tasks.HandleMessage),
 		RetryLimit:      3,
 		MinBackoff:      time.Second,
 	})
@@ -81,13 +81,14 @@ func (q *Queue) initAddQueue() {
 }
 
 func (q *Queue) initDelQueue() {
+	queueName := "azsqs:" + q.opt.Name + ":delete"
 	q.delQueue = memqueue.NewQueue(&taskq.QueueOptions{
-		Name:       "azsqs:" + q.opt.Name + ":delete",
+		Name:       queueName,
 		BufferSize: 100,
 		Redis:      q.opt.Redis,
 	})
-	q.delTask = q.delQueue.NewTask(&taskq.TaskOptions{
-		Name:       "delete-message",
+	q.delTask = taskq.NewTask(&taskq.TaskOptions{
+		Name:       queueName + ":delete-message",
 		Handler:    taskq.HandlerFunc(q.delBatcherAdd),
 		RetryLimit: 3,
 		MinBackoff: time.Second,
@@ -110,22 +111,6 @@ func (q *Queue) Options() *taskq.QueueOptions {
 	return q.opt
 }
 
-func (q *Queue) HandleMessage(msg *taskq.Message) error {
-	return q.base.HandleMessage(msg)
-}
-
-func (q *Queue) NewTask(opt *taskq.TaskOptions) *taskq.Task {
-	return q.base.NewTask(q, opt)
-}
-
-func (q *Queue) GetTask(name string) *taskq.Task {
-	return q.base.GetTask(name)
-}
-
-func (q *Queue) RemoveTask(name string) {
-	q.base.RemoveTask(name)
-}
-
 func (q *Queue) Consumer() *taskq.Consumer {
 	if q.consumer == nil {
 		q.consumer = taskq.NewConsumer(q)
@@ -152,8 +137,11 @@ func (q *Queue) Add(msg *taskq.Message) error {
 	if msg.TaskName == "" {
 		return internal.ErrTaskNameRequired
 	}
+	if q.isDuplicate(msg) {
+		return taskq.ErrDuplicate
+	}
 	msg = msgutil.WrapMessage(msg)
-	return q.addTask.AddMessage(msg)
+	return q.addQueue.Add(q.addTask.WithMessage(msg))
 }
 
 func (q *Queue) queueURL() string {
@@ -286,7 +274,8 @@ func (q *Queue) Release(msg *taskq.Message) error {
 
 // Delete deletes the message from the queue.
 func (q *Queue) Delete(msg *taskq.Message) error {
-	return q.delTask.AddMessage(msgutil.WrapMessage(msg))
+	msg = msgutil.WrapMessage(msg)
+	return q.delQueue.Add(q.delTask.WithMessage(msg))
 }
 
 // Purge deletes all messages from the queue using SQS API.
@@ -368,8 +357,8 @@ func (q *Queue) addBatch(msgs []*taskq.Message) error {
 		}
 
 		if len(str) > msgSizeLimit {
-			internal.Logger.Printf("%s: str=%d bytes=%d is larger than %d",
-				msg.Task, len(str), len(b), msgSizeLimit)
+			internal.Logger.Printf("task=%q: str=%d bytes=%d is larger than %d",
+				msg.TaskName, len(str), len(b), msgSizeLimit)
 		}
 
 		entry := &sqs.SendMessageBatchRequestEntry{
@@ -520,6 +509,13 @@ func (q *Queue) GetDeleteQueue() *memqueue.Queue {
 	return q.delQueue
 }
 
+func (q *Queue) isDuplicate(msg *taskq.Message) bool {
+	if msg.Name == "" {
+		return false
+	}
+	return q.opt.Storage.Exists("taskq:" + q.opt.Name + ":" + msg.Name)
+}
+
 func findMessageById(msgs []*taskq.Message, id string) *taskq.Message {
 	i, err := strconv.Atoi(id)
 	if err != nil {

bench_test.go

@@ -19,18 +19,19 @@ func BenchmarkConsumerRedisq(b *testing.B) {
 
 var (
 	once sync.Once
+	q    taskq.Queuer
 	task *taskq.Task
 	wg   sync.WaitGroup
 )
 
 func benchmarkConsumer(b *testing.B, factory taskq.Factory) {
 	once.Do(func() {
-		q := factory.NewQueue(&taskq.QueueOptions{
+		q = factory.NewQueue(&taskq.QueueOptions{
 			Name:  "bench",
 			Redis: redisRing(),
 		})
 
-		task = q.NewTask(&taskq.TaskOptions{
+		task = taskq.NewTask(&taskq.TaskOptions{
 			Name: "bench",
 			Handler: func() {
 				wg.Done()
@@ -45,7 +46,7 @@ func benchmarkConsumer(b *testing.B, factory taskq.Factory) {
 	for i := 0; i < b.N; i++ {
 		for j := 0; j < 100; j++ {
 			wg.Add(1)
-			_ = task.Call()
+			_ = q.Add(task.WithArgs())
 		}
 		wg.Wait()
 	}