bugfixes

Author: idanasulin2706

README.md

@@ -238,8 +238,7 @@ curl --location --request POST 'rest_gateway:4444/stations/<station_name>/consum
     "consumer_name": <consumer_name> string required,
     "consumer_group": <consumer_group> string defaults to <consumer_name>,
     "batch_size": <batch_size> integer defaults to 10,
-    "batch_max_wait_time_ms": <batch_max_wait_time> integer defaults to 5 secs,
-    "max_msg_deliveries": <max_msg_deliveries> integer defaults to 10
+    "batch_max_wait_time_ms": <batch_max_wait_time> integer defaults to 5 secs
 }'
 ```
 
@@ -250,49 +249,15 @@ Expected output:
   {
     "message": "{\n    \"message\": \"message x\"\n}",
     "headers": {
-      "$memphis_connectionId": "ab083953-a3b1-467b-8487-dfcd24e2ba59",
-      "$memphis_producedBy": "rest_gateway",
       "Accept": "*/*",
-      "Accept-Encoding": "gzip, deflate, br",
-      "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2NvdW50X2lkIjoxLCJjb25uZWN0aW9uX3Rva2VuIjoiIiwiZXhwIjoxNjkzNzc2MzYzLCJwYXNzd29yZCI6IkBlYnViZUBBZ3UxIiwidXNlcm5hbWUiOiJlYnViZWFndSJ9.QW2sx1mVyyQ1g88Gclj9VNfwvXU4Fr__M1XXbjc6jis",
-      "Connection": "keep-alive",
-      "Content-Length": "30",
-      "Content-Type": "application/json",
-      "Host": "localhost:4444",
-      "Postman-Token": "1c1947d9-5dfb-4c3a-8da7-a31fbf993be6",
-      "User-Agent": "PostmanRuntime/7.32.3"
+      "Accept-Encoding": "gzip, deflate, br"
     }
   },
   {
     "message": "{\n    \"message\": \"message y\"\n}",
     "headers": {
-      "$memphis_connectionId": "ab083953-a3b1-467b-8487-dfcd24e2ba59",
-      "$memphis_producedBy": "rest_gateway",
-      "Accept": "*/*",
-      "Accept-Encoding": "gzip, deflate, br",
-      "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2NvdW50X2lkIjoxLCJjb25uZWN0aW9uX3Rva2VuIjoiIiwiZXhwIjoxNjkzNzc2MzYzLCJwYXNzd29yZCI6IkBlYnViZUBBZ3UxIiwidXNlcm5hbWUiOiJlYnViZWFndSJ9.QW2sx1mVyyQ1g88Gclj9VNfwvXU4Fr__M1XXbjc6jis",
-      "Connection": "keep-alive",
-      "Content-Length": "30",
-      "Content-Type": "application/json",
-      "Host": "localhost:4444",
-      "Postman-Token": "9f76af3e-a861-483b-a2f0-b3f7e9e72125",
-      "User-Agent": "PostmanRuntime/7.32.3"
-    }
-  },
-  {
-    "message": "{\n    \"message\": \"Hello world\"\n}",
-    "headers": {
-      "$memphis_connectionId": "ab083953-a3b1-467b-8487-dfcd24e2ba59",
-      "$memphis_producedBy": "rest_gateway",
-      "Accept": "*/*",
-      "Accept-Encoding": "gzip, deflate, br",
-      "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2NvdW50X2lkIjoxLCJjb25uZWN0aW9uX3Rva2VuIjoiIiwiZXhwIjoxNjkzNzc2MzYzLCJwYXNzd29yZCI6IkBlYnViZUBBZ3UxIiwidXNlcm5hbWUiOiJlYnViZWFndSJ9.QW2sx1mVyyQ1g88Gclj9VNfwvXU4Fr__M1XXbjc6jis",
-      "Connection": "keep-alive",
-      "Content-Length": "32",
       "Content-Type": "application/json",
-      "Host": "localhost:4444",
-      "Postman-Token": "7fe95b24-edae-4415-bc4c-258e6c93a51d",
-      "User-Agent": "PostmanRuntime/7.32.3"
+      "Host": "localhost:4444"
     }
   }
 ]

handlers/consumer.go

@@ -2,36 +2,35 @@ package handlers
 
 import (
 	"fmt"
-	"github.com/gofiber/fiber/v2"
-	"github.com/memphisdev/memphis.go"
 	"rest-gateway/logger"
 	"rest-gateway/models"
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/gofiber/fiber/v2"
+	"github.com/memphisdev/memphis.go"
 )
 
 type requestBody struct {
 	ConsumerName       string `json:"consumer_name"`
 	ConsumerGroup      string `json:"consumer_group"`
 	BatchSize          int    `json:"batch_size"`
 	BatchMaxWaitTimeMs int    `json:"batch_max_wait_time_ms"`
-	MaxMsgDeliveries   int    `json:"max_msg_deliveries"`
 }
 
 func (r *requestBody) initializeDefaults() {
 	if r.ConsumerGroup == "" {
-		r.ConsumerGroup = r.ConsumerName
+		r.ConsumerGroup = "rest-gateway"
+	} else {
+		r.ConsumerGroup = fmt.Sprintf("%s-rest-gateway", r.ConsumerGroup)
 	}
 	if r.BatchSize == 0 {
 		r.BatchSize = 10
 	}
 	if r.BatchMaxWaitTimeMs == 0 {
 		r.BatchMaxWaitTimeMs = 5000
 	}
-	if r.MaxMsgDeliveries == 0 {
-		r.MaxMsgDeliveries = 10
-	}
 }
 
 func ConsumeHandleMessage() func(*fiber.Ctx) error {
@@ -71,9 +70,7 @@ func ConsumeHandleMessage() func(*fiber.Ctx) error {
 		accountIdStr := strconv.Itoa(int(accountId))
 		conn := ConnectionsCache[accountIdStr][username].Connection
 		if conn == nil {
-			errMsg := fmt.Sprintf("Connection does not exist")
-			log.Errorf("ConsumeHandleMessage - consume: %s", errMsg)
-
+			log.Warnf("ConsumeHandleMessage - consume: Connection does not exist")
 			c.Status(fiber.StatusInternalServerError)
 			return c.JSON(&fiber.Map{
 				"success": false,
@@ -85,14 +82,14 @@ func ConsumeHandleMessage() func(*fiber.Ctx) error {
 			memphis.FetchBatchSize(reqBody.BatchSize),
 			memphis.FetchConsumerGroup(reqBody.ConsumerGroup),
 			memphis.FetchBatchMaxWaitTime(time.Duration(reqBody.BatchMaxWaitTimeMs)*time.Millisecond),
-			memphis.FetchMaxMsgDeliveries(reqBody.MaxMsgDeliveries))
+			memphis.FetchMaxMsgDeliveries(1))
 
 		if err != nil {
 			log.Errorf("ConsumeHandleMessage - fetch messages: %s", err.Error())
-			c.Status(fiber.StatusInternalServerError)
+			c.Status(fiber.StatusBadRequest)
 			return c.JSON(&fiber.Map{
 				"success": false,
-				"error":   "Server error",
+				"error":   err.Error(),
 			})
 		}
 

handlers/producer.go

@@ -3,7 +3,6 @@ package handlers
 import (
 	"encoding/json"
 	"errors"
-	"fmt"
 	"strconv"
 
 	"rest-gateway/logger"
@@ -72,16 +71,14 @@ func CreateHandleMessage() func(*fiber.Ctx) error {
 			accountIdStr := strconv.Itoa(int(accountId))
 			conn := ConnectionsCache[accountIdStr][username].Connection
 			if conn == nil {
-				errMsg := fmt.Sprintf("Connection does not exist")
-				log.Errorf("CreateHandleMessage - produce: %s", errMsg)
-
+				log.Warnf("CreateHandleMessage - produce: Connection does not exist")
 				c.Status(fiber.StatusInternalServerError)
 				return c.JSON(&fiber.Map{
 					"success": false,
 					"error":   "Server error",
 				})
 			}
-			err = conn.Produce(stationName, "rest_gateway", message, []memphis.ProducerOpt{}, []memphis.ProduceOpt{memphis.MsgHeaders(hdrs)})
+			err = conn.Produce(stationName, "rest-gateway", message, []memphis.ProducerOpt{}, []memphis.ProduceOpt{memphis.MsgHeaders(hdrs)})
 			if err != nil {
 				log.Errorf("CreateHandleMessage - produce: %s", err.Error())
 				c.Status(fiber.StatusInternalServerError)
@@ -147,9 +144,7 @@ func CreateHandleBatch() func(*fiber.Ctx) error {
 			accountIdStr := strconv.Itoa(int(accountId))
 			conn := ConnectionsCache[accountIdStr][username].Connection
 			if conn == nil {
-				errMsg := fmt.Sprintf("Connection does not exist")
-				log.Errorf("CreateHandleBatch - produce: %s", errMsg)
-
+				log.Warnf("CreateHandleBatch - produce: Connection does not exist")
 				c.Status(fiber.StatusInternalServerError)
 				return c.JSON(&fiber.Map{
 					"success": false,
@@ -166,7 +161,7 @@ func CreateHandleBatch() func(*fiber.Ctx) error {
 					allErr = append(allErr, err.Error())
 					continue
 				}
-				if err := conn.Produce(stationName, "rest_gateway", rawRes, []memphis.ProducerOpt{}, []memphis.ProduceOpt{memphis.MsgHeaders(hdrs)}); err != nil {
+				if err := conn.Produce(stationName, "rest-gateway", rawRes, []memphis.ProducerOpt{}, []memphis.ProduceOpt{memphis.MsgHeaders(hdrs)}); err != nil {
 					log.Errorf("CreateHandleBatch - produce: %s", err.Error())
 					errCount++
 					allErr = append(allErr, err.Error())