chore: update task message handling and dependencies

- Update `github.com/golang-queue/queue` dependency from v0.2.1 to v0.3.0
- Remove indirect dependencies `github.com/vmihailenco/msgpack/v5` and `github.com/vmihailenco/tagparser/v2`
- Change `QueuedMessage` to `TaskMessage` in function signatures in `options.go`
- Change `QueuedMessage` to `TaskMessage` in function signatures in `redis.go`
- Change `QueuedMessage` to `TaskMessage` in test function signatures in `redis_test.go`

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

go.mod

@@ -4,7 +4,7 @@ go 1.22
 
 require (
 	github.com/appleboy/com v0.2.1
-	github.com/golang-queue/queue v0.2.1
+	github.com/golang-queue/queue v0.3.0
 	github.com/redis/go-redis/v9 v9.7.0
 	github.com/stretchr/testify v1.10.0
 	github.com/testcontainers/testcontainers-go v0.35.0
@@ -53,8 +53,6 @@ require (
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
-	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
-	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.3 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
 	go.opentelemetry.io/otel v1.24.0 // indirect

go.sum

@@ -50,8 +50,8 @@ github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
-github.com/golang-queue/queue v0.2.1 h1:W0JeHniILD4oxvs+E/hDuV0zlnQh0qeTy73BAjrigCw=
-github.com/golang-queue/queue v0.2.1/go.mod h1:eUZ3HH9GbhoEKQSlxCBQ4pPXeadbJ7QKBMZ0kIZNDHI=
+github.com/golang-queue/queue v0.3.0 h1:gyBLNT9EDOsChazYScp8iLiwLfG0SdnCDmNUybcHig4=
+github.com/golang-queue/queue v0.3.0/go.mod h1:SkjMwz1TjxZOrF7kABvbar1CagcMxwRtXt5Tx00wb4g=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
@@ -125,10 +125,6 @@ github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFA
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
 github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
 github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
-github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
-github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
-github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
-github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=

options.go

@@ -13,7 +13,7 @@ import (
 type Option func(*options)
 
 type options struct {
-	runFunc          func(context.Context, core.QueuedMessage) error
+	runFunc          func(context.Context, core.TaskMessage) error
 	logger           queue.Logger
 	addr             string
 	db               int
@@ -110,7 +110,7 @@ func WithConnectionString(connectionString string) Option {
 }
 
 // WithRunFunc setup the run func of queue
-func WithRunFunc(fn func(context.Context, core.QueuedMessage) error) Option {
+func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option {
 	return func(w *options) {
 		w.runFunc = fn
 	}
@@ -157,7 +157,7 @@ func newOptions(opts ...Option) options {
 		group:      "golang-queue",
 		consumer:   "golang-queue",
 		logger:     queue.NewLogger(),
-		runFunc: func(context.Context, core.QueuedMessage) error {
+		runFunc: func(context.Context, core.TaskMessage) error {
 			return nil
 		},
 		blockTime: 60 * time.Second,

redis.go

@@ -176,7 +176,7 @@ func (w *Worker) queue(data interface{}) error {
 }
 
 // Queue send notification to queue
-func (w *Worker) Queue(task core.QueuedMessage) error {
+func (w *Worker) Queue(task core.TaskMessage) error {
 	if atomic.LoadInt32(&w.stopFlag) == 1 {
 		return queue.ErrQueueShutdown
 	}
@@ -185,12 +185,12 @@ func (w *Worker) Queue(task core.QueuedMessage) error {
 }
 
 // Run start the worker
-func (w *Worker) Run(ctx context.Context, task core.QueuedMessage) error {
+func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error {
 	return w.opts.runFunc(ctx, task)
 }
 
 // Request a new task
-func (w *Worker) Request() (core.QueuedMessage, error) {
+func (w *Worker) Request() (core.TaskMessage, error) {
 	clock := 0
 	w.startConsumer()
 loop:

redis_test.go

@@ -141,7 +141,7 @@ func TestCustomFuncAndWait(t *testing.T) {
 	w := NewWorker(
 		WithAddr(endpoint),
 		WithStreamName("test3"),
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+		WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			time.Sleep(500 * time.Millisecond)
 			return nil
 		}),
@@ -189,7 +189,7 @@ func TestRedisCluster(t *testing.T) {
 		WithAddr(strings.Join(hosts, ",")),
 		WithStreamName("testCluster"),
 		WithCluster(),
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+		WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			time.Sleep(500 * time.Millisecond)
 			return nil
 		}),
@@ -243,7 +243,7 @@ func TestJobReachTimeout(t *testing.T) {
 	w := NewWorker(
 		WithAddr(endpoint),
 		WithStreamName("timeout"),
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+		WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			for {
 				select {
 				case <-ctx.Done():
@@ -286,7 +286,7 @@ func TestCancelJobAfterShutdown(t *testing.T) {
 		WithAddr(endpoint),
 		WithStreamName("cancel"),
 		WithLogger(queue.NewLogger()),
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+		WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			for {
 				select {
 				case <-ctx.Done():
@@ -329,7 +329,7 @@ func TestGoroutineLeak(t *testing.T) {
 		WithAddr(endpoint),
 		WithStreamName("GoroutineLeak"),
 		WithLogger(queue.NewEmptyLogger()),
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+		WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			for {
 				select {
 				case <-ctx.Done():
@@ -376,7 +376,7 @@ func TestGoroutinePanic(t *testing.T) {
 	w := NewWorker(
 		WithAddr(endpoint),
 		WithStreamName("GoroutinePanic"),
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+		WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			panic("missing something")
 		}),
 	)