fix support application/x-protobuf

Author: shohamroditimemphis

handlers/producer.go

@@ -3,116 +3,83 @@ package handler
 import (
 	"encoding/json"
 	"errors"
-	"log"
 	"strings"
 
 	"github.com/gofiber/fiber/v2"
 	"github.com/memphisdev/memphis.go"
-	"google.golang.org/protobuf/proto"
 )
 
-type Handler struct{ P *memphis.Producer }
+func handleHeaders(headers map[string]string) (memphis.Headers, error) {
+	hdrs := memphis.Headers{}
+	hdrs.New()
 
-func (p Handler) produce(message []byte, hdrs memphis.Headers) error {
-	if err := p.P.Produce(message, memphis.MsgHeaders(hdrs)); err != nil {
-		return err
+	for key, value := range headers {
+		err := hdrs.Add(key, value)
+		if err != nil {
+			return memphis.Headers{}, err
+		}
 	}
-	return nil
-
+	return hdrs, nil
 }
 
-func handleMessageHdrs(bodyReq []byte, hdrs memphis.Headers) ([]byte, memphis.Headers, error) {
+func handleJsonMessage(bodyReq []byte, headers map[string]string) ([]byte, memphis.Headers, error) {
 	type body struct {
 		Message string `json:"message"`
-		Headers string `json:"headers"`
 	}
-	var b body
-	err := json.Unmarshal(bodyReq, &b)
+	var bodyRequest body
+	err := json.Unmarshal(bodyReq, &bodyRequest)
 	if err != nil {
 		return nil, memphis.Headers{}, err
 	}
 
-	var headers map[string]string
-	err = json.Unmarshal([]byte(b.Headers), &headers)
+	hdrs, err := handleHeaders(headers)
 	if err != nil {
 		return nil, memphis.Headers{}, err
 	}
 
-	var k, v string
-	for key, value := range headers {
-		k = key
-		v = value
-
-		err = hdrs.Add(k, v)
-		if err != nil {
-			return nil, memphis.Headers{}, err
-		}
-	}
-
-	message, err := json.Marshal(b.Message)
+	message, err := json.Marshal(bodyRequest.Message)
 	if err != nil {
 		return nil, memphis.Headers{}, err
 	}
 	return message, hdrs, nil
 }
 
-func CreateHandleMessage(p *memphis.Producer) func(*fiber.Ctx) error {
+func CreateHandleMessage(producer *memphis.Producer) func(*fiber.Ctx) error {
 	return func(c *fiber.Ctx) error {
 		bodyReq := c.Body()
+		headers := c.GetReqHeaders()
 		contentType := string(c.Request().Header.ContentType())
 		var message []byte
-		var err error
 		hdrs := memphis.Headers{}
-		hdrs.New()
+		var err error
 		caseText := strings.Contains(contentType, "text")
-
 		if caseText {
 			contentType = "text/"
 		}
 
 		switch contentType {
 		case "application/json":
-			message, hdrs, err = handleMessageHdrs(bodyReq, hdrs)
+			message, hdrs, err = handleJsonMessage(bodyReq, headers)
 			if err != nil {
 				return err
 			}
 		case "text/":
-			message, hdrs, err = handleMessageHdrs(bodyReq, hdrs)
+			message = bodyReq
+			hdrs, err = handleHeaders(headers)
 			if err != nil {
 				return err
 			}
 		case "application/x-protobuf":
-			msg := &Msg{}
-			err := proto.Unmarshal(bodyReq, msg)
-			if err != nil {
-				log.Fatal("unmarshaling error: ", err)
-			}
-
-			message, err = json.Marshal(msg.Message)
+			message = bodyReq
+			hdrs, err = handleHeaders(headers)
 			if err != nil {
 				return err
 			}
-
-			var headers map[string]string
-			err = json.Unmarshal([]byte(msg.Headers), &headers)
-			if err != nil {
-				return err
-			}
-
-			var k, v string
-			for key, value := range headers {
-				k = key
-				v = value
-				err = hdrs.Add(k, v)
-				if err != nil {
-					return err
-				}
-			}
 		default:
 			return errors.New("unsupported content type")
 		}
 
-		if err := p.Produce(message, memphis.MsgHeaders(hdrs)); err != nil {
+		if err := producer.Produce(message, memphis.MsgHeaders(hdrs)); err != nil {
 			c.Status(400)
 			return c.JSON(&fiber.Map{
 				"success": false,