Change NatsMessenger to Singleton Init

This commit is contained in:
Christoph Kluge 2024-08-13 10:33:41 +02:00
parent 0bf316dbf4
commit 940f3e7bcd
2 changed files with 60 additions and 67 deletions

View File

@ -216,10 +216,7 @@ func main() {
// Start NATS Messenger if Config exists // Start NATS Messenger if Config exists
wg.Add(1) wg.Add(1)
nm, err := natsMessenger.New(config.Keys.Nats) nm := natsMessenger.GetNatsMessenger(config.Keys.Nats)
if err != nil {
log.Fatal("Error on NATS startup!")
}
wg.Done() wg.Done()
// Start HTTP server // Start HTTP server

View File

@ -12,6 +12,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"os" "os"
"sync"
"time" "time"
"github.com/ClusterCockpit/cc-backend/internal/importer" "github.com/ClusterCockpit/cc-backend/internal/importer"
@ -31,9 +32,8 @@ type NatsMessenger struct {
jwtPubKey ed25519.PublicKey jwtPubKey ed25519.PublicKey
} }
func New(config *schema.NatsConfig) (nm *NatsMessenger, err error) { var natsMessengerInstance *NatsMessenger
return SetupNatsMessenger(config) var once sync.Once
}
type DevNatsMessage struct { type DevNatsMessage struct {
Content string `json:"content"` Content string `json:"content"`
@ -72,75 +72,71 @@ type ReceiveEventNatsRequest struct {
Value *int64 `json:"value,omitempty" example:"150"` // Optional Value Set for Evenr, eg powercap Value *int64 `json:"value,omitempty" example:"150"` // Optional Value Set for Evenr, eg powercap
} }
// Check auth and setup listeners to channels // Get Singleton
func GetNatsMessenger(config *schema.NatsConfig) *NatsMessenger {
// ns *server.Server, nc *nats.Conn, subs []*nats.Subscription, err error
func SetupNatsMessenger(config *schema.NatsConfig) (nm *NatsMessenger, err error) {
// Check if Config present // Check if Config present
if config == nil { if config == nil {
log.Info("No NATS config found: Skip NATS init.") log.Info("No NATS config found: Skip NATS init.")
return nil, nil return nil
} }
// Init Raw if natsMessengerInstance == nil {
nmr := NatsMessenger{ once.Do(
func() {
// Raw Init
var err error
natsMessengerInstance = &NatsMessenger{
Server: nil, Server: nil,
Connection: nil, Connection: nil,
Subscriptions: []*nats.Subscription{}, Subscriptions: []*nats.Subscription{},
JobRepository: repository.GetJobRepository(), JobRepository: repository.GetJobRepository(),
jwtPubKey: nil, jwtPubKey: nil,
} }
// Init JWT PubKey // Init JWT PubKey
pubKey := os.Getenv("JWT_PUBLIC_KEY") pubKey := os.Getenv("JWT_PUBLIC_KEY")
if pubKey == "" { if pubKey == "" {
log.Warn("environment variable 'JWT_PUBLIC_KEY' not set (token based authentication will not work for nats: abort setup)") log.Warn("environment variable 'JWT_PUBLIC_KEY' not set (token based authentication will not work for nats: abort setup)")
return nil, fmt.Errorf("environment variable 'JWT_PUBLIC_KEY' not set (token based authentication will not work for nats: abort nats setup)")
} else { } else {
bytes, err := base64.StdEncoding.DecodeString(pubKey) if bytes, err := base64.StdEncoding.DecodeString(pubKey); err != nil {
if err != nil {
log.Warn("Could not decode JWT public key") log.Warn("Could not decode JWT public key")
return nil, err } else {
natsMessengerInstance.jwtPubKey = ed25519.PublicKey(bytes)
} }
nmr.jwtPubKey = ed25519.PublicKey(bytes)
} }
// Start Nats Server // Start Nats Server
// Note: You can configure things like Host, Port, Authorization, and much more using server.Options. // Note: You can configure things like Host, Port, Authorization, and much more using server.Options.
opts := &server.Options{Port: config.Port} opts := &server.Options{Port: config.Port}
nmr.Server, err = server.NewServer(opts) if natsMessengerInstance.Server, err = server.NewServer(opts); err != nil {
if err != nil {
log.Error("nats server error on creation") log.Error("nats server error on creation")
return nil, err
} }
go nmr.Server.Start() go natsMessengerInstance.Server.Start()
if !nmr.Server.ReadyForConnections(3 * time.Second) { if !natsMessengerInstance.Server.ReadyForConnections(3 * time.Second) {
log.Error("nats server not ready for connection") log.Error("nats server not ready for connection")
return nil, fmt.Errorf("nats server not ready for connection")
} }
// Connect // Connect
var copts []nats.Option var copts []nats.Option
nmr.Connection, err = nats.Connect(nmr.Server.ClientURL(), copts...) if natsMessengerInstance.Connection, err = nats.Connect(natsMessengerInstance.Server.ClientURL(), copts...); err != nil {
if nmr.Connection == nil { natsMessengerInstance.Server.Shutdown()
nmr.Server.Shutdown()
log.Error("nats connection could not be established: nats shut down") log.Error("nats connection could not be established: nats shut down")
return nil, err
} }
// Subscribe // Subscribe
if err = nmr.setupSubscriptions(); err != nil { if err = natsMessengerInstance.setupSubscriptions(); err != nil {
log.Error("error when subscribing to channels: nats shut down") log.Error("error when subscribing to channels: nats shut down")
nmr.Connection.Close() natsMessengerInstance.Connection.Close()
nmr.Server.Shutdown() natsMessengerInstance.Server.Shutdown()
return nil, err }
})
log.Infof("NATS server and subscriptions on port '%d' established\n", config.Port)
} else {
log.Infof("Single NatsMessenger instance already created on port '%d'\n", config.Port)
} }
log.Infof("NATS server and subscriptions on port '%d' established\n", config.Port) return natsMessengerInstance
return &nmr, nil
} }
func (nm *NatsMessenger) StopNatsMessenger() { func (nm *NatsMessenger) StopNatsMessenger() {