apply FatalOnConnectionClose method to each rbmq connection in nocloud

slntopp · Dec 21, 2024 · 28cc5b5 · 28cc5b5
1 parent f646cf1
commit 28cc5b5
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 11 deletions.
diff --git a/cmd/billing/main.go b/cmd/billing/main.go
@@ -165,6 +165,7 @@ func main() {
 	}
 	_, _ = ch.QueueDelete(".", false, false, false)
 	ch.Close()
+	rabbitmq.FatalOnConnectionClose(log, conn)
 
 	// Initialize controllers
 	accountsCtrl := graph.NewAccountsController(log, db)

diff --git a/cmd/edge/main.go b/cmd/edge/main.go
@@ -95,13 +95,14 @@ func main() {
 	)
 
 	log.Info("Dialing RabbitMQ", zap.String("url", rbmq))
-	rbmq, err := amqp.Dial(rbmq)
+	conn, err := amqp.Dial(rbmq)
 	if err != nil {
 		log.Fatal("Failed to connect to RabbitMQ", zap.Error(err))
 	}
-	defer rbmq.Close()
+	defer conn.Close()
+	rabbitmq.FatalOnConnectionClose(log, conn)
 
-	server := edge.NewEdgeServiceServer(log, db, rabbitmq.NewRabbitMQConnection(rbmq))
+	server := edge.NewEdgeServiceServer(log, db, rabbitmq.NewRabbitMQConnection(conn))
 	pb.RegisterEdgeServiceServer(s, server)
 
 	healthpb.RegisterInternalProbeServiceServer(s, NewHealthServer(log))

diff --git a/cmd/eventbus/main.go b/cmd/eventbus/main.go
@@ -77,6 +77,7 @@ func main() {
 		log.Fatal("failed to connect to RabbitMQ", zap.Error(err))
 	}
 	defer conn.Close()
+	rabbitmq.FatalOnConnectionClose(log, conn)
 	log.Info("RabbitMQ connection established")
 
 	rdb := redis.NewClient(&redis.Options{

diff --git a/cmd/services_registry/main.go b/cmd/services_registry/main.go
@@ -139,11 +139,12 @@ func main() {
 	})
 
 	log.Info("Dialing RabbitMQ", zap.String("url", rbmq))
-	rbmq, err := amqp.Dial(rbmq)
+	conn, err := amqp.Dial(rbmq)
 	if err != nil {
 		log.Fatal("Failed to connect to RabbitMQ", zap.Error(err))
 	}
-	defer rbmq.Close()
+	defer conn.Close()
+	rabbitmq.FatalOnConnectionClose(log, conn)
 
 	log.Info("Setting up Pub/Sub")
 	ps, err := states.SetupStatesStreaming()
@@ -152,8 +153,8 @@ func main() {
 	}
 	log.Info("Pub/Sub setted up")
 
-	server := services.NewServicesServer(log, db, ps, rabbitmq.NewRabbitMQConnection(rbmq))
-	iserver := instances.NewInstancesServiceServer(log, db, rabbitmq.NewRabbitMQConnection(rbmq), rdb)
+	server := services.NewServicesServer(log, db, ps, rabbitmq.NewRabbitMQConnection(conn))
+	iserver := instances.NewInstancesServiceServer(log, db, rabbitmq.NewRabbitMQConnection(conn), rdb)
 
 	for _, driver := range drivers {
 		log.Info("Registering Driver", zap.String("driver", driver))
@@ -248,7 +249,7 @@ func main() {
 	ctx = context.WithValue(ctx, nocloud.NoCloudAccount, schema.ROOT_ACCOUNT_KEY)
 	ctx, cancel := context.WithCancel(ctx)
 	go iserver.MonitoringRoutine(ctx, worker(workers))
-	_ps := pubsub.NewPubSub[*epb.Event](rabbitmq.NewRabbitMQConnection(rbmq), log)
+	_ps := pubsub.NewPubSub[*epb.Event](rabbitmq.NewRabbitMQConnection(conn), log)
 	go iserver.ConsumeInvokeCommands(log, ctx, _ps, worker(workers))
 
 	host := fmt.Sprintf("0.0.0.0:%s", port)

diff --git a/cmd/sp_registry/main.go b/cmd/sp_registry/main.go
@@ -107,14 +107,15 @@ func main() {
 	)
 
 	log.Info("Dialing RabbitMQ", zap.String("url", rbmq))
-	rbmq, err := amqp.Dial(rbmq)
+	conn, err := amqp.Dial(rbmq)
 	if err != nil {
 		log.Fatal("Failed to connect to RabbitMQ", zap.Error(err))
 	}
-	defer rbmq.Close()
+	defer conn.Close()
+	rabbitmq.FatalOnConnectionClose(log, conn)
 	log.Info("RabbitMQ connection established")
 
-	server := sp.NewServicesProviderServer(log, db, rabbitmq.NewRabbitMQConnection(rbmq), rdb)
+	server := sp.NewServicesProviderServer(log, db, rabbitmq.NewRabbitMQConnection(conn), rdb)
 	s_server := showcases.NewShowcasesServer(log, db)
 
 	log.Debug("Got drivers", zap.Strings("drivers", drivers))