update the endpoint

Author: shohamroditimemphis

handlers/producer.go

@@ -44,14 +44,29 @@ func handleJsonMessage(bodyReq []byte, headers map[string]string) ([]byte, memph
 	return message, hdrs, nil
 }
 
-func CreateHandleMessage(producer *memphis.Producer) func(*fiber.Ctx) error {
+func CreateHandleMessage(conn *memphis.Conn) func(*fiber.Ctx) error {
+	producers := make(map[string]*memphis.Producer)
 	return func(c *fiber.Ctx) error {
+		stationName := c.Params("stationName")
+		producerName := c.Params("producerName")
+		var producer *memphis.Producer
+		var err error
+
+		if len(producers) == 0 || producers[stationName].Name != producerName {
+			producer, err = conn.CreateProducer(stationName, producerName)
+			if err != nil {
+				return err
+			}
+			producers[stationName] = producer
+		} else {
+			producer = producers[stationName]
+		}
+
 		bodyReq := c.Body()
 		headers := c.GetReqHeaders()
 		contentType := string(c.Request().Header.ContentType())
 		var message []byte
 		hdrs := memphis.Headers{}
-		var err error
 		caseText := strings.Contains(contentType, "text")
 		if caseText {
 			contentType = "text/"

main.go

@@ -20,12 +20,7 @@ func main() {
 		fmt.Println(err.Error())
 		os.Exit(1)
 	}
-	producer, err := conn.CreateProducer("test-fiber-go", "simple_go_producer")
-	if err != nil {
-		fmt.Println(err.Error())
-		os.Exit(1)
-	}
 
-	router.SetupRoutes(app, producer)
+	router.SetupRoutes(app, conn)
 	app.Listen(":3000")
 }

router/router.go

@@ -8,8 +8,7 @@ import (
 )
 
 // SetupRoutes setup router api
-func SetupRoutes(app *fiber.App, producer *memphis.Producer) {
-
+func SetupRoutes(app *fiber.App, conn *memphis.Conn) {
 	api := app.Group("/", logger.New())
-	api.Post("/", handler.CreateHandleMessage(producer))
+	api.Post("/station/:stationName/:producerName", handler.CreateHandleMessage(conn))
 }