Merge pull request #7 from frain-dev/feat/msg-args-bin

add marshall functions for message

Author: ogbanugot

brokers/redis/list.go

@@ -174,7 +174,7 @@ func (b *List) Publish(msg *disq.Message) error {
 		msg.ID = uuid.NewString()
 	}
 
-	body, err := msgpack.Marshal((*disq.MessageRaw)(msg))
+	body, err := msg.MarshalBinary()
 	if err != nil {
 		return err
 	}

brokers/redis/stream.go

@@ -184,7 +184,7 @@ func (b *Stream) Publish(msg *disq.Message) error {
 		msg.ID = uuid.NewString()
 	}
 
-	body, err := msgpack.Marshal((*disq.MessageRaw)(msg))
+	body, err := msg.MarshalBinary()
 	if err != nil {
 		return err
 	}

go.mod

@@ -6,7 +6,6 @@ require (
 	github.com/bsm/redislock v0.7.2
 	github.com/go-redis/redis/v8 v8.11.4
 	github.com/google/uuid v1.3.0
-	github.com/hashicorp/golang-lru v0.5.4
 	github.com/sirupsen/logrus v1.8.1
 	github.com/vmihailenco/msgpack v4.0.4+incompatible
 	github.com/vmihailenco/msgpack/v5 v5.3.5

go.sum

@@ -36,8 +36,6 @@ github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
-github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=

message.go

@@ -3,6 +3,8 @@ package disq
 import (
 	"context"
 	"time"
+
+	"github.com/vmihailenco/msgpack"
 )
 
 // Message is used as a uniform object for publishing and consuming messages from a queue.
@@ -18,8 +20,8 @@ type Message struct {
 	Delay time.Duration
 
 	// Args passed to the handler.
-	Args []interface{}
-
+	Args    []interface{}
+	ArgsBin []byte
 	// The number of times the message has been reserved or released.
 	RetryCount int
 
@@ -36,6 +38,27 @@ func NewMessage(ctx context.Context, args ...interface{}) *Message {
 	}
 }
 
+func (m *Message) MarshalBinary() ([]byte, error) {
+	bArgs, err := msgpack.Marshal(m.Args)
+	if err != nil {
+		return nil, err
+	}
+	m.ArgsBin = bArgs
+
+	bMsg, err := msgpack.Marshal((*MessageRaw)(m))
+	if err != nil {
+		return nil, err
+	}
+	return bMsg, nil
+}
+
+func (m *Message) UnmarshalBinary(b []byte) error {
+	if err := msgpack.Unmarshal(b, (*MessageRaw)(m)); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (m *Message) SetDelay(delay time.Duration) {
 	m.Delay = delay
 }