Use Message.TaskName when checking message uniqueness

Author: vmihailenco

azsqs/queue.go

@@ -510,7 +510,7 @@ func (q *Queue) isDuplicate(msg *taskq.Message) bool {
 	if msg.Name == "" {
 		return false
 	}
-	return q.opt.Storage.Exists("taskq:" + q.opt.Name + ":" + msg.Name)
+	return q.opt.Storage.Exists(msgutil.FullMessageName(q, msg))
 }
 
 func findMessageById(msgs []*taskq.Message, id string) *taskq.Message {

internal/msgutil/msgutil.go

@@ -39,3 +39,7 @@ func UnwrapMessageHandler(fn interface{}) taskq.HandlerFunc {
 		return h(msg)
 	})
 }
+
+func FullMessageName(q taskq.Queue, msg *taskq.Message) string {
+	return "taskq:" + q.Name() + ":" + msg.TaskName + ":" + msg.Name
+}

ironmq/queue.go

@@ -284,7 +284,7 @@ func (q *Queue) isDuplicate(msg *taskq.Message) bool {
 	if msg.Name == "" {
 		return false
 	}
-	return q.opt.Storage.Exists("taskq:" + q.opt.Name + ":" + msg.Name)
+	return q.opt.Storage.Exists(msgutil.FullMessageName(q, msg))
 }
 
 func retry(fn func() error) error {

memqueue/queue.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/vmihailenco/taskq/v2"
 	"github.com/vmihailenco/taskq/v2/internal"
+	"github.com/vmihailenco/taskq/v2/internal/msgutil"
 )
 
 type Queue struct {
@@ -173,5 +174,5 @@ func (q *Queue) isDuplicate(msg *taskq.Message) bool {
 	if msg.Name == "" {
 		return false
 	}
-	return q.opt.Storage.Exists("taskq:" + q.opt.Name + ":" + msg.Name)
+	return q.opt.Storage.Exists(msgutil.FullMessageName(q, msg))
 }

message.go

@@ -1,9 +1,11 @@
 package taskq
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"time"
 
 	"github.com/valyala/gozstd"
@@ -74,6 +76,28 @@ func (m *Message) OnceInPeriod(period time.Duration, args ...interface{}) *Messa
 	return m
 }
 
+func hashArgs(args []interface{}) []byte {
+	var buf bytes.Buffer
+	enc := msgpack.NewEncoder(&buf)
+	_ = enc.EncodeMulti(args...)
+	b := buf.Bytes()
+
+	if len(b) <= 32 {
+		return b
+	}
+
+	h := fnv.New128a()
+	_, _ = h.Write(b)
+	return h.Sum(nil)
+}
+
+func timeSlot(period time.Duration) int64 {
+	if period <= 0 {
+		return 0
+	}
+	return time.Now().UnixNano() / int64(period)
+}
+
 func (m *Message) MarshalArgs() ([]byte, error) {
 	if m.ArgsBin != nil {
 		if m.ArgsCompression == "" {

redisq/queue.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/vmihailenco/taskq/v2"
 	"github.com/vmihailenco/taskq/v2/internal"
+	"github.com/vmihailenco/taskq/v2/internal/msgutil"
 	"github.com/vmihailenco/taskq/v2/internal/redislock"
 )
 
@@ -374,7 +375,7 @@ func (q *Queue) isDuplicate(msg *taskq.Message) bool {
 	if msg.Name == "" {
 		return false
 	}
-	return q.opt.Storage.Exists("taskq:" + q.opt.Name + ":" + msg.Name)
+	return q.opt.Storage.Exists(msgutil.FullMessageName(q, msg))
 }
 
 func (q *Queue) withRedisLock(name string, fn func() error) error {

task.go

@@ -1,13 +1,9 @@
 package taskq
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"hash/fnv"
 	"time"
-
-	"github.com/vmihailenco/msgpack/v4"
 )
 
 var unknownTaskOpt *TaskOptions
@@ -121,25 +117,3 @@ func (t *Task) WithArgs(ctx context.Context, args ...interface{}) *Message {
 	msg.TaskName = t.opt.Name
 	return msg
 }
-
-func timeSlot(period time.Duration) int64 {
-	if period <= 0 {
-		return 0
-	}
-	return time.Now().UnixNano() / int64(period)
-}
-
-func hashArgs(args []interface{}) []byte {
-	var buf bytes.Buffer
-	enc := msgpack.NewEncoder(&buf)
-	_ = enc.EncodeMulti(args...)
-	b := buf.Bytes()
-
-	if len(b) <= 64 {
-		return b
-	}
-
-	h := fnv.New128a()
-	_, _ = h.Write(b)
-	return h.Sum(nil)
-}