Merge pull request #7 from memphisdev/batch-message-support

Batch message support

Author: idanasulin2706

handlers/producer.go

@@ -1,7 +1,9 @@
 package handler
 
 import (
+	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 
 	"github.com/gofiber/fiber/v2"
@@ -83,3 +85,63 @@ func CreateHandleMessage(conn *memphis.Conn) func(*fiber.Ctx) error {
 		})
 	}
 }
+
+func CreateHandleBatch(conn *memphis.Conn) func(*fiber.Ctx) error {
+	return func(c *fiber.Ctx) error {
+		stationName := c.Params("stationName")
+		var producer *memphis.Producer
+
+		producer, err := createProducer(conn, producers, stationName)
+		if err != nil {
+			return err
+		}
+
+		bodyReq := c.Body()
+		headers := c.GetReqHeaders()
+		contentType := string(c.Request().Header.ContentType())
+
+		switch contentType {
+		case "application/json":
+			var batchReq []map[string]string
+			err := json.Unmarshal(bodyReq, &batchReq)
+			if err != nil {
+				return errors.New("unsupported request")
+			}
+			hdrs, err := handleHeaders(headers)
+			if err != nil {
+				return err
+			}
+
+			errCount := 0
+			var lastErr error
+			for _, msg := range batchReq {
+				rawRes, err := json.Marshal(msg)
+				if err != nil {
+					errCount++
+					lastErr = err
+					continue
+				}
+				if err := producer.Produce(rawRes, memphis.MsgHeaders(hdrs)); err != nil {
+					errCount++
+					lastErr = err
+				}
+			}
+
+			if errCount > 0 {
+				c.Status(400)
+				return c.JSON(&fiber.Map{
+					"success": false,
+					"error":   fmt.Sprintf("send failed for %d/%d messages, last error: %v", errCount, len(batchReq), lastErr.Error()),
+				})
+			}
+		default:
+			return errors.New("unsupported content type")
+		}
+
+		c.Status(200)
+		return c.JSON(&fiber.Map{
+			"success": true,
+			"error":   nil,
+		})
+	}
+}

router/router.go

@@ -1,14 +1,16 @@
 package router
 
 import (
+	handler "http-proxy/handlers"
+
 	"github.com/gofiber/fiber/v2"
 	"github.com/gofiber/fiber/v2/middleware/logger"
 	"github.com/memphisdev/memphis.go"
-	handler "http-proxy/handlers"
 )
 
 // SetupRoutes setup router api
 func SetupRoutes(app *fiber.App, conn *memphis.Conn) {
 	api := app.Group("/stations", logger.New())
 	api.Post("/:stationName/produce/single", handler.CreateHandleMessage(conn))
+	api.Post("/:stationName/produce/batch", handler.CreateHandleBatch(conn))
 }