make consume method to auto acknowledge the messages and no equivalent acknowledge method for the gateway

Author: EbubeCode

README.md

@@ -217,6 +217,54 @@ Expected output:
 ```json
 {"errors":["Schema validation has failed: jsonschema: '' does not validate with file:///Users/user/memphisdev/memphis-rest-gateway/123#/required: missing properties: 'field1'","Schema validation has failed: jsonschema: '' does not validate with file:///Users/user/memphisdev/memphis-rest-gateway/123#/required: missing properties: 'field1'"],"fail":2,"sent":1,"success":false}
 ```
+### Consume a batch of messages 
+
+Attach the JWT token to every request.\
+JWT token as '`Bearer`' as a header.
+
+The messages are auto acknowledged by the rest gateway.
+
+#### Supported content types:
+
+* application/json
+
+#### Example:
+
+```bash
+curl --location --request POST 'rest_gateway:4444/stations/<station_name>/consume/batch' \
+--header 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.e30.4KOGRhUaqvm-qSHnmMwX5VrLKsvHo33u3UdJ0qYP0kI' \
+--header 'Content-Type: application/json' \
+--data-raw '{
+    "consumer_name": <consumer_name> string required,
+    "consumer_group_name": <consumer_group_name> 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,
+    "max_msg_deliveries": <max_msg_deliveries> integer defaults to 10
+}'
+```
+
+Expected output:
+
+```json
+[
+    {
+        "message": "{\n    \"message\": \"How're you doing\"\n}"
+    },
+    {
+        "message": "{\n    \"message\": \"How far\"\n}"
+    }
+]
+```
+
+#### Error Examples:
+
+```json
+{
+    "error": "Consumer name is required",
+    "success": false
+}
+```
 
 ## Support 🙋‍♂️🤝
 

handlers/consumer.go

@@ -20,7 +20,7 @@ type requestBody struct {
 	MaxMsgDeliveries  int    `json:"max_msg_deliveries"`
 }
 
-func (r requestBody) initializeDefaults() {
+func (r *requestBody) initializeDefaults() {
 	if r.ConsumerGroupName == "" {
 		r.ConsumerGroupName = r.ConsumerName
 	}
@@ -88,28 +88,28 @@ func ConsumeHandleMessage() func(*fiber.Ctx) error {
 		msgs, err := conn.FetchMessages(stationName, reqBody.ConsumerName,
 			memphis.FetchBatchSize(reqBody.BatchSize),
 			memphis.FetchConsumerGroup(reqBody.ConsumerGroupName),
-			memphis.FetchBatchMaxWaitTime(time.Duration(reqBody.BatchMaxWaitTime)),
-			memphis.FetchMaxAckTime(time.Duration(reqBody.MaxAckTime)),
+			memphis.FetchBatchMaxWaitTime(time.Duration(reqBody.BatchMaxWaitTime)*time.Second),
+			memphis.FetchMaxAckTime(time.Duration(reqBody.MaxAckTime)*time.Second),
 			memphis.FetchMaxMsgDeliveries(reqBody.MaxMsgDeliveries))
 
 		if err != nil {
 			log.Errorf("ConsumeHandleMessage - fetch messages: %s", err.Error())
-			c.Status(fiber.StatusBadRequest)
+			c.Status(fiber.StatusInternalServerError)
 			return c.JSON(&fiber.Map{
 				"success": false,
 				"error":   "Server error",
 			})
 		}
 
 		type message struct {
-			Data string `json:"data"`
+			Message string `json:"message"`
 		}
 		messages := []message{}
 
 		for _, msg := range msgs {
 			err := msg.Ack()
 			if err != nil {
-				log.Errorf("ConsumeHandleMessage - consume: %s", err)
+				log.Errorf("ConsumeHandleMessage - acknowledge message: %s", err)
 			}
 			messages = append(messages, message{string(msg.Data())})
 		}

main.go

@@ -18,7 +18,7 @@ func initializeLogger() *logger.Logger {
 			creds := configuration.CONNECTION_TOKEN
 			username := configuration.ROOT_USER
 			if configuration.USER_PASS_BASED_AUTH {
-				username = "$$memphis"
+				username = "$memphis"
 				creds = configuration.CONNECTION_TOKEN + "_" + configuration.ROOT_PASSWORD
 			}
 			l, err := logger.CreateLogger(configuration.MEMPHIS_HOST, username, creds)

router/stations.go

@@ -11,5 +11,5 @@ func InitializeStationsRoutes(app *fiber.App) {
 	api := app.Group("/stations", logger.New())
 	api.Post("/:stationName/produce/single", handlers.CreateHandleMessage())
 	api.Post("/:stationName/produce/batch", handlers.CreateHandleBatch())
-	api.Get("/:stationName/consume/batch", handlers.ConsumeHandleMessage())
+	api.Post("/:stationName/consume/batch", handlers.ConsumeHandleMessage())
 }