implement message acknowledgement

Author: EbubeCode

cache/cache.go

@@ -33,15 +33,16 @@ func New(config conf.Configuration, log *logger.Logger) MessageCache {
 		db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
 		if err != nil {
 			log.Errorf("MessageCache.New - %s", err.Error())
-			log.Noticef("Defaulting to in-memory cache")
 		} else {
 			repo := Repository{db}
 			err := repo.MigrateMessages()
 			if err != nil {
 				log.Errorf("MessageCache.New - %s", err.Error())
 			}
+			log.Noticef("Using database for message cache")
 			return repo
 		}
 	}
+	log.Noticef("Defaulting to in-memory cache")
 	return InMemoryCache{map[string]map[string][]Message{}}
 }

conf/config.go

@@ -28,7 +28,7 @@ type Configuration struct {
 	DB_PASSWORD                    string
 	DB_NAME                        string
 	DB_PORT                        string
-	DB_SSLMODE                     bool
+	DB_SSLMODE                     string
 	DB_TIME_ZONE                   string
 }
 

handlers/consumer.go

@@ -185,7 +185,7 @@ func getQueryParamValue(urlString, paramName string) (string, error) {
 }
 
 func getConsumerNameParameterValue(urlString, defaultValue string) string {
-	consumerName, err := getQueryParamValue(urlString, "consumerName")
+	consumerName, err := getQueryParamValue(urlString, "consumer_name")
 	if err != nil {
 		consumerName = defaultValue
 	}
@@ -194,7 +194,7 @@ func getConsumerNameParameterValue(urlString, defaultValue string) string {
 }
 
 func getBatchSizeParameterValue(urlString string, defaultValue int) int {
-	batchSizeString, _ := getQueryParamValue(urlString, "batchSize")
+	batchSizeString, _ := getQueryParamValue(urlString, "batch_size")
 	batchSize, err := strconv.ParseInt(batchSizeString, 10, 64)
 	if err != nil {
 		return defaultValue

main.go

@@ -37,7 +37,6 @@ func main() {
 	l := initalizeLogger()
 	go handlers.CleanConnectionsCache()
 	app := router.SetupRoutes(l)
-	handlers.InitializeMessageCache(l)
 	l.Noticef("Memphis REST gateway is up and running")
 	l.Noticef("Version %s", configuration.VERSION)
 	app.Listen(":" + configuration.HTTP_PORT)

router/router.go

@@ -1,6 +1,7 @@
 package router
 
 import (
+	"rest-gateway/handlers"
 	"rest-gateway/logger"
 	"rest-gateway/middlewares"
 	"rest-gateway/utils"
@@ -15,6 +16,7 @@ func SetupRoutes(l *logger.Logger) *fiber.App {
 	app := fiber.New(fiber.Config{
 		DisableStartupMessage: true,
 	})
+	handlers.InitializeMessageCache(l)
 
 	logger.SetLogger(app, l)
 	app.Use(cors.New())

router/stations.go

@@ -11,4 +11,6 @@ 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", handlers.ConsumeHandleMessage())
+	api.Post("/:stationName/ack", handlers.AcknowledgeMessage())
 }