code review

Author: EbubeCode

README.md

@@ -236,10 +236,9 @@ curl --location --request POST 'rest_gateway:4444/stations/<station_name>/consum
 --header 'Content-Type: application/json' \
 --data-raw '{
     "consumer_name": <consumer_name> string required,
-    "consumer_group_name": <consumer_group_name> string defaults to <consumer_name>,
+    "consumer_group": <consumer_group> string defaults to <consumer_name>,
     "batch_size": <batch_size> integer defaults to 10,
-    "batch_max_wait_time": <batch_max_wait_time> integer defaults to 5 secs,
-    "max_ack_time": <max_ack_time> integer defaults to 30 secs,
+    "batch_max_wait_time_ms": <batch_max_wait_time> integer defaults to 5 secs,
     "max_msg_deliveries": <max_msg_deliveries> integer defaults to 10
 }'
 ```
@@ -248,12 +247,54 @@ Expected output:
 
 ```json
 [
-    {
-        "message": "{\n    \"message\": \"How're you doing\"\n}"
-    },
-    {
-        "message": "{\n    \"message\": \"How far\"\n}"
+  {
+    "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"
     }
+  },
+  {
+    "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"
+    }
+  }
 ]
 ```
 

handlers/consumer.go

@@ -12,26 +12,22 @@ import (
 )
 
 type requestBody struct {
-	ConsumerName      string `json:"consumer_name"`
-	ConsumerGroupName string `json:"consumer_group_name"`
-	BatchSize         int    `json:"batch_size"`
-	MaxAckTime        int    `json:"max_ack_time"`
-	BatchMaxWaitTime  int    `json:"batch_max_wait_time"`
-	MaxMsgDeliveries  int    `json:"max_msg_deliveries"`
+	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.ConsumerGroupName == "" {
-		r.ConsumerGroupName = r.ConsumerName
+	if r.ConsumerGroup == "" {
+		r.ConsumerGroup = r.ConsumerName
 	}
 	if r.BatchSize == 0 {
 		r.BatchSize = 10
 	}
-	if r.BatchMaxWaitTime == 0 {
-		r.BatchMaxWaitTime = 5
-	}
-	if r.MaxAckTime == 0 {
-		r.MaxAckTime = 30
+	if r.BatchMaxWaitTimeMs == 0 {
+		r.BatchMaxWaitTimeMs = 5000
 	}
 	if r.MaxMsgDeliveries == 0 {
 		r.MaxMsgDeliveries = 10
@@ -87,9 +83,8 @@ func ConsumeHandleMessage() func(*fiber.Ctx) error {
 		reqBody.initializeDefaults()
 		msgs, err := conn.FetchMessages(stationName, reqBody.ConsumerName,
 			memphis.FetchBatchSize(reqBody.BatchSize),
-			memphis.FetchConsumerGroup(reqBody.ConsumerGroupName),
-			memphis.FetchBatchMaxWaitTime(time.Duration(reqBody.BatchMaxWaitTime)*time.Second),
-			memphis.FetchMaxAckTime(time.Duration(reqBody.MaxAckTime)*time.Second),
+			memphis.FetchConsumerGroup(reqBody.ConsumerGroup),
+			memphis.FetchBatchMaxWaitTime(time.Duration(reqBody.BatchMaxWaitTimeMs)*time.Millisecond),
 			memphis.FetchMaxMsgDeliveries(reqBody.MaxMsgDeliveries))
 
 		if err != nil {
@@ -102,7 +97,8 @@ func ConsumeHandleMessage() func(*fiber.Ctx) error {
 		}
 
 		type message struct {
-			Message string `json:"message"`
+			Message string            `json:"message"`
+			Headers map[string]string `json:"headers"`
 		}
 		messages := []message{}
 
@@ -111,7 +107,10 @@ func ConsumeHandleMessage() func(*fiber.Ctx) error {
 			if err != nil {
 				log.Errorf("ConsumeHandleMessage - acknowledge message: %s", err)
 			}
-			messages = append(messages, message{string(msg.Data())})
+			messages = append(messages, message{
+				Message: string(msg.Data()),
+				Headers: msg.GetHeaders(),
+			})
 		}
 		c.Status(fiber.StatusOK)
 		return c.JSON(&messages)