Merge pull request #10 from frain-dev/enhancement/msgpack-encoding

add reflect decoding

Author: ogbanugot

brokers/redis/list.go

@@ -253,7 +253,8 @@ func (b *List) Status() bool {
 }
 
 func ListUnmarshalMessage(msg *disq.Message, body string) error {
-	if err := msgpack.Unmarshal([]byte(body), (*disq.MessageRaw)(msg)); err != nil {
+	err := msg.UnmarshalBinary(disq.StringToBytes(body))
+	if err != nil {
 		return err
 	}
 	return nil

brokers/redis/stream.go

@@ -13,7 +13,6 @@ import (
 	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
-	"github.com/vmihailenco/msgpack"
 )
 
 // Broker based on redis STREAM and ZSET.
@@ -316,7 +315,8 @@ func unixMs(tm time.Time) int64 {
 
 func StreamUnmarshalMessage(msg *disq.Message, xmsg *redis.XMessage) error {
 	body := xmsg.Values["body"].(string)
-	if err := msgpack.Unmarshal([]byte(body), (*disq.MessageRaw)(msg)); err != nil {
+	err := msg.UnmarshalBinary(disq.StringToBytes(body))
+	if err != nil {
 		return err
 	}
 	msg.ID = xmsg.ID

example/api.go

@@ -72,16 +72,26 @@ func NewWorker(c *redis.Client, brokers []disq.Broker) *Worker {
 	}
 }
 
-//create task
-var CountHandler, _ = disq.RegisterTask(&disq.TaskOptions{
-	Name: "CountHandler",
-	Handler: func(name string) error {
+type MsgValue struct {
+	Name  string
+	Value string
+}
+
+func counthandler() func(job *MsgValue) error {
+	return func(msgval *MsgValue) error {
 		//time.Sleep(time.Duration(10) * time.Second)
-		// fmt.Println("Hello", name)
+		fmt.Println("Hello", msgval.Value)
 		// return errors.New("error")
 		// return &EndpointError{Err: errors.New("Error"), delay: time.Second * 60}
 		return nil
-	},
+	}
+
+}
+
+//create task
+var CountHandler, _ = disq.RegisterTask(&disq.TaskOptions{
+	Name:       "CountHandler",
+	Handler:    counthandler(),
 	RetryLimit: 3,
 })
 

example/publisher/publisher.go

@@ -11,8 +11,7 @@ import (
 )
 
 func main() {
-
-	count := 50000000
+	count := 50
 	go func() {
 		for i := 0; i < count; i++ {
 			value := fmt.Sprint("message_", uuid.NewString())
@@ -21,7 +20,10 @@ func main() {
 			msg := &disq.Message{
 				Ctx:      ctx,
 				TaskName: example.CountHandler.Name(),
-				Args:     []interface{}{value},
+				Args: []interface{}{&example.MsgValue{
+					Name:  "test",
+					Value: value,
+				}},
 				// Delay:    delay,
 			}
 			err := example.RWorker.Worker.Brokers()[0].Publish(msg)
@@ -41,7 +43,10 @@ func main() {
 				msg := &disq.Message{
 					Ctx:      ctx,
 					TaskName: example.CountHandler.Name(),
-					Args:     []interface{}{value},
+					Args: []interface{}{&example.MsgValue{
+						Name:  "test",
+						Value: value,
+					}},
 					// Delay:    delay,
 				}
 				err := example.RWorker.Worker.Brokers()[1].Publish(msg)
@@ -64,7 +69,10 @@ func main() {
 				msg := &disq.Message{
 					Ctx:      ctx,
 					TaskName: example.CountHandler.Name(),
-					Args:     []interface{}{value},
+					Args: []interface{}{&example.MsgValue{
+						Name:  "test",
+						Value: value,
+					}},
 					// Delay:    delay,
 				}
 				err := example.RWorker.Worker.Brokers()[2].Publish(msg)

handler.go

@@ -137,6 +137,12 @@ func (h *reflectFunc) fnArgs(msg *Message) ([]reflect.Value, error) {
 
 	for i := 0; i < len(in); i++ {
 		arg := reflect.New(h.ft.In(inStart + i)).Elem()
+		err = dec.DecodeValue(arg)
+		if err != nil {
+			err = fmt.Errorf(
+				"disq: decoding arg=%d failed (data=%.100x): %s", i, b, err)
+			return nil, err
+		}
 		in[i] = arg
 	}
 

util.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"strconv"
 	"time"
+	"unsafe"
 )
 
 func UnixMs(tm time.Time) int64 {
@@ -21,3 +22,18 @@ func ConsumerName() string {
 func DurEqual(d1, d2 time.Duration, threshold int) bool {
 	return (d2 >= d1 && (d2-d1) < time.Duration(threshold)*time.Second)
 }
+
+// BytesToString converts byte slice to string.
+func BytesToString(b []byte) string {
+	return *(*string)(unsafe.Pointer(&b))
+}
+
+// StringToBytes converts string to byte slice.
+func StringToBytes(s string) []byte {
+	return *(*[]byte)(unsafe.Pointer(
+		&struct {
+			string
+			Cap int
+		}{s, len(s)},
+	))
+}