Memphis nats integration tcp

analytics/analytics.go

@@ -43,8 +43,8 @@ var disableAnalyticsCounter metric.Int64Counter
 var deploymentId string
 var analyticsFlag string
 
-func InitializeAnalytics() error {
-	systemKeysCollection = db.GetCollection("system_keys")
+func InitializeAnalytics(c *mongo.Client) error {
+	systemKeysCollection = db.GetCollection("system_keys", c)
 	deployment, err := getSystemKey("deployment_id")
 	if err == mongo.ErrNoDocuments {
 		deploymentId := primitive.NewObjectID().Hex()
@@ -90,6 +90,7 @@ func InitializeAnalytics() error {
 		return err
 	} else {
 		analyticsFlag = analytics.Value
+
 	}
 
 	ls = launcher.ConfigureOpentelemetry(

background_tasks/poison_messages_listener.go

@@ -14,11 +14,10 @@
 package background_tasks
 
 import (
-	"memphis-broker/handlers"
 	"memphis-broker/server"
 )
 
-var poisonMessagesHandler handlers.PoisonMessagesHandler
+var poisonMessagesHandler server.PoisonMessagesHandler
 
 func ListenForPoisonMessages(s *server.Server) {
 	s.QueueSubscribe("$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>",

background_tasks/zombie_resources.go

@@ -36,10 +36,10 @@ var poisonMessagesCollection *mongo.Collection
 var serv *server.Server
 
 func InitializeZombieResources(s *server.Server) {
-	connectionsCollection = db.GetCollection("connections")
-	producersCollection = db.GetCollection("producers")
-	consumersCollection = db.GetCollection("consumers")
-	poisonMessagesCollection = db.GetCollection("poison_messages")
+	connectionsCollection = db.GetCollection("connections", s.DbClient)
+	producersCollection = db.GetCollection("producers", s.DbClient)
+	consumersCollection = db.GetCollection("consumers", s.DbClient)
+	poisonMessagesCollection = db.GetCollection("poison_messages", s.DbClient)
 	serv = s
 }
 

db/db.go

@@ -15,7 +15,6 @@ package db
 
 import (
 	"memphis-broker/conf"
-	"memphis-broker/server"
 
 	"context"
 	"time"
@@ -25,13 +24,24 @@ import (
 )
 
 var configuration = conf.GetConfig()
-var serv *server.Server
+// var serv *server.Server
 
 const (
 	dbOperationTimeout = 20
 )
 
-func InitializeDbConnection(s *server.Server) error {
+type logger interface{
+	Noticef(string, ...interface{})
+	Errorf(string, ...interface{})
+}
+
+type DbInstance struct {
+	Client  *mongo.Client
+	Ctx	context.Context
+	Cancel context.CancelFunc
+} 
+
+func InitializeDbConnection(l logger) (DbInstance, error) {
 	ctx, cancel := context.WithTimeout(context.TODO(), dbOperationTimeout*time.Second)
 
 	var clientOptions *options.ClientOptions
@@ -48,32 +58,34 @@ func InitializeDbConnection(s *server.Server) error {
 
 	client, err := mongo.Connect(ctx, clientOptions)
 	if err != nil {
-		return err
+		cancel()
+		return DbInstance{}, err
 	}
 
 	err = client.Ping(ctx, nil)
 	if err != nil {
-		return err
+		cancel()
+		return DbInstance{}, err
 	}
 
-	s.DbClient = client
-	s.DbCtx = ctx
-	s.DbCancel = cancel
-	serv = s
-	s.Noticef("Established connection with the DB")
-	return nil
+	// s.DbClient = client
+	// s.DbCtx = ctx
+	// s.DbCancel = cancel
+	// serv = s
+	l.Noticef("Established connection with the DB")
+	return DbInstance{Client:client, Ctx: ctx, Cancel: cancel}, nil
 }
 
-func GetCollection(collectionName string) *mongo.Collection {
-	var collection *mongo.Collection = serv.DbClient.Database(configuration.DB_NAME).Collection(collectionName)
+func GetCollection(collectionName string, dbClient *mongo.Client) *mongo.Collection {
+	var collection *mongo.Collection = dbClient.Database(configuration.DB_NAME).Collection(collectionName)
 	return collection
 }
 
-func Close() {
-	defer serv.DbCancel()
+func Close(dbi DbInstance, l logger) {
+	defer dbi.Cancel()
 	defer func() {
-		if err := serv.DbClient.Disconnect(serv.DbCtx); err != nil {
-			serv.Errorf("Failed to close Mongodb client: " + err.Error())
+		if err := dbi.Client.Disconnect(dbi.Ctx); err != nil {
+			l.Errorf("Failed to close Mongodb client: " + err.Error())
 		}
 	}()
 }

http_server/http_server.go

@@ -15,7 +15,6 @@ package http_server
 
 import (
 	"memphis-broker/conf"
-	"memphis-broker/handlers"
 	"memphis-broker/http_server/routes"
 	"memphis-broker/server"
 	"memphis-broker/socketio"
@@ -25,14 +24,14 @@ import (
 func InitializeHttpServer(s *server.Server, wg *sync.WaitGroup) {
 	configuration := conf.GetConfig()
 
-	handlers := handlers.Handlers{
-		Producers:  handlers.ProducersHandler{S: s},
-		Consumers:  handlers.ConsumersHandler{S: s},
-		AuditLogs:  handlers.AuditLogsHandler{},
-		Stations:   handlers.StationsHandler{S: s},
-		Factories:  handlers.FactoriesHandler{S: s},
-		Monitoring: handlers.MonitoringHandler{S: s},
-		PoisonMsgs: handlers.PoisonMessagesHandler{S: s},
+	handlers := server.Handlers{
+		Producers:  server.ProducersHandler{S: s},
+		Consumers:  server.ConsumersHandler{S: s},
+		AuditLogs:  server.AuditLogsHandler{},
+		Stations:   server.StationsHandler{S: s},
+		Factories:  server.FactoriesHandler{S: s},
+		Monitoring: server.MonitoringHandler{S: s},
+		PoisonMsgs: server.PoisonMessagesHandler{S: s},
 	}
 
 	httpServer := routes.InitializeHttpRoutes(&handlers)

http_server/routes/consumers.go

@@ -14,12 +14,12 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
+	"memphis-broker/server"
 
 	"github.com/gin-gonic/gin"
 )
 
-func InitializeConsumersRoutes(router *gin.RouterGroup, h *handlers.Handlers) {
+func InitializeConsumersRoutes(router *gin.RouterGroup, h *server.Handlers) {
 	consumersHandler := h.Consumers
 	consumersRoutes := router.Group("/consumers")
 	consumersRoutes.GET("/getAllConsumers", consumersHandler.GetAllConsumers)

http_server/routes/factories.go

@@ -14,12 +14,12 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
+	"memphis-broker/server"
 
 	"github.com/gin-gonic/gin"
 )
 
-func InitializeFactoriesRoutes(router *gin.RouterGroup, h *handlers.Handlers) {
+func InitializeFactoriesRoutes(router *gin.RouterGroup, h *server.Handlers) {
 	factoriesHandler := h.Factories
 	factoriesRoutes := router.Group("/factories")
 	factoriesRoutes.POST("/createFactory", factoriesHandler.CreateFactory)

http_server/routes/monitoring.go

@@ -14,12 +14,12 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
+	"memphis-broker/server"
 
 	"github.com/gin-gonic/gin"
 )
 
-func InitializeMonitoringRoutes(router *gin.RouterGroup, h *handlers.Handlers) {
+func InitializeMonitoringRoutes(router *gin.RouterGroup, h *server.Handlers) {
 	monitoringHandler := h.Monitoring
 	monitoringRoutes := router.Group("/monitoring")
 	monitoringRoutes.GET("/getClusterInfo", monitoringHandler.GetClusterInfo)

http_server/routes/producers.go

@@ -14,12 +14,12 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
+	"memphis-broker/server"
 
 	"github.com/gin-gonic/gin"
 )
 
-func InitializeProducersRoutes(router *gin.RouterGroup, h *handlers.Handlers) {
+func InitializeProducersRoutes(router *gin.RouterGroup, h *server.Handlers) {
 	producersHandler := h.Producers
 	producersRoutes := router.Group("/producers")
 	producersRoutes.GET("/getAllProducers", producersHandler.GetAllProducers)

http_server/routes/routes.go

@@ -14,15 +14,15 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
 	"memphis-broker/middlewares"
+	"memphis-broker/server"
 	"memphis-broker/utils"
 
 	"github.com/gin-contrib/cors"
 	"github.com/gin-gonic/gin"
 )
 
-func InitializeHttpRoutes(handlers *handlers.Handlers) *gin.Engine {
+func InitializeHttpRoutes(handlers *server.Handlers) *gin.Engine {
 	router := gin.Default()
 
 	router.Use(cors.New(cors.Config{

http_server/routes/sandbox.go

@@ -14,13 +14,13 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
+	"memphis-broker/server"
 
 	"github.com/gin-gonic/gin"
 )
 
 func InitializeSandboxRoutes(router *gin.RouterGroup) {
-	sandboxHandler := handlers.SandboxHandler{}
+	sandboxHandler := server.SandboxHandler{}
 	sandboxRoutes := router.Group("/sandbox")
 	sandboxRoutes.POST("/login", sandboxHandler.Login)
 }

http_server/routes/stations.go

@@ -14,12 +14,12 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
+	"memphis-broker/server"
 
 	"github.com/gin-gonic/gin"
 )
 
-func InitializeStationsRoutes(router *gin.RouterGroup, h *handlers.Handlers) {
+func InitializeStationsRoutes(router *gin.RouterGroup, h *server.Handlers) {
 	stationsHandler := h.Stations
 	stationsRoutes := router.Group("/stations")
 	stationsRoutes.GET("/getStation", stationsHandler.GetStation)

http_server/routes/user_mgmt.go

@@ -14,13 +14,12 @@
 package routes
 
 import (
-	"memphis-broker/handlers"
-
+	"memphis-broker/server"
 	"github.com/gin-gonic/gin"
 )
 
 func InitializeUserMgmtRoutes(router *gin.RouterGroup) {
-	userMgmtHandler := handlers.UserMgmtHandler{}
+	userMgmtHandler := server.UserMgmtHandler{}
 	userMgmtRoutes := router.Group("/usermgmt")
 	userMgmtRoutes.GET("/nats/authenticate", userMgmtHandler.AuthenticateNatsUser)
 	userMgmtRoutes.GET("/nats/authenticate/:publicKey", userMgmtHandler.AuthenticateNatsUser)

main.go

@@ -21,7 +21,6 @@ import (
 	"memphis-broker/analytics"
 	"memphis-broker/background_tasks"
 	"memphis-broker/db"
-	"memphis-broker/handlers"
 	"memphis-broker/http_server"
 	"memphis-broker/server"
 	"os"
@@ -93,48 +92,51 @@ func usage() {
 	os.Exit(0)
 }
 
-func handleError(s *server.Server, message string, err error) {
-	if err != nil {
-		s.Errorf(message + " " + err.Error())
-	}
-}
-
 func runMemphis(s *server.Server) {
 
 	if !s.MemphisInitialized() {
 		s.Fatalf("Jetstream not enabled on global account")
 	}
 
-	err := db.InitializeDbConnection(s)
-	handleError(s, "Failed initializing db connection: ", err)
-
-	err = analytics.InitializeAnalytics()
-	handleError(s, "Failed initializing analytics: ", err)
+	dbInstance, err := db.InitializeDbConnection(s)
+	if err != nil {
+		s.Errorf("Failed initializing db connection:" + " " + err.Error())
+		os.Exit(1)
+	}
 
-	handlers.InitializeHandlers(s)
+	err = analytics.InitializeAnalytics(dbInstance.Client)
+	if err != nil {
+		s.Errorf("Failed initializing analytics: " + " " + err.Error())
+		db.Close(dbInstance, s)
+		os.Exit(1)
+	}
 
-	err = handlers.CreateRootUserOnFirstSystemLoad()
-	handleError(s, "Failed to create root user: ", err)
+	s.InitializeMemphisHandlers(dbInstance)
 
+	err = server.CreateRootUserOnFirstSystemLoad()
+	if err != nil {
+		s.Errorf("Failed to create root user: " + " " + err.Error())
+		db.Close(dbInstance, s)
+		os.Exit(1)
+	}
 	background_tasks.InitializeZombieResources(s)
 
-	defer db.Close()
+	defer db.Close(dbInstance, s)
 
 	// defer broker.Close()
 	defer analytics.Close()
 
 	wg := new(sync.WaitGroup)
 	wg.Add(4)
 
-	// go tcp_server.InitializeTcpServer(wg)
 	go http_server.InitializeHttpServer(s, wg)
 	go background_tasks.KillZombieResources(wg)
 	go background_tasks.ListenForPoisonMessages(s)
 
 	var env string
 	if os.Getenv("DOCKER_ENV") != "" {
 		env = "Docker"
-		s.Noticef("\n**********\n\nDashboard: http://localhost:9000\nMemphis broker: localhost:5555 (Management Port) / 7766 (Data Port) / 6666 (TCP Port)\nUI/CLI root username - root\nUI/CLI root password - memphis\nSDK root connection token - memphis  \n\n**********")
+		s.Noticef("\n**********\n\nDashboard: http://localhost:9000\nMemphis broker: localhost:5555 (Management Port) / 6666 (TCP Port)\nUI/CLI root username - root\nUI/CLI root password - memphis\nSDK root connection token - memphis  \n\n**********")
 	} else {
 		env = "K8S"
 	}