add reflect decoding

Author: ogbanugot

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
 	}