diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 518ac51a2..9bda35ca2 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -541,25 +541,11 @@ func (d *Daemon) Start() error { // connectNATS establishes a connection to the NATS server with reconnect handling. func (d *Daemon) connectNATS() error { - opts := []nats.Option{ - nats.Token(d.config.NATS.ACL.Token), - nats.ReconnectWait(time.Second), - nats.MaxReconnects(-1), // Infinite reconnects - nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { - slog.Warn("NATS disconnected", "err", err) - }), - nats.ReconnectHandler(func(nc *nats.Conn) { - slog.Info("NATS reconnected", "url", nc.ConnectedUrl()) - }), - } - - var err error - d.natsConn, err = nats.Connect(d.config.NATS.Host, opts...) + nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token) if err != nil { return err } - - slog.Info("Connected to NATS server", "host", d.config.NATS.Host) + d.natsConn = nc return nil } diff --git a/hive/services/awsgw/awsgw.go b/hive/services/awsgw/awsgw.go index 12b31ba80..197a32017 100644 --- a/hive/services/awsgw/awsgw.go +++ b/hive/services/awsgw/awsgw.go @@ -5,12 +5,10 @@ import ( "log/slog" "os" "strings" - "time" "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/gateway" "github.com/mulgadc/hive/hive/utils" - "github.com/nats-io/nats.go" ) var serviceName = "awsgw" @@ -60,27 +58,13 @@ func launchService(config *config.ClusterConfig) (err error) { nodeConfig := config.Nodes[config.Node] // Connect to NATS for service communication - opts := []nats.Option{ - nats.Token(nodeConfig.NATS.ACL.Token), - nats.ReconnectWait(time.Second), - nats.MaxReconnects(-1), // Infinite reconnects - nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { - slog.Warn("NATS disconnected", "err", err) - }), - nats.ReconnectHandler(func(nc *nats.Conn) { - slog.Info("NATS reconnected", "url", nc.ConnectedUrl()) - }), - } - - natsConn, err := nats.Connect(nodeConfig.NATS.Host, opts...) + natsConn, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token) if err != nil { slog.Error("Failed to connect to NATS", "err", err) return err } defer natsConn.Close() - slog.Info("Connected to NATS server", "host", nodeConfig.NATS.Host) - // Append Base dir if config has no leading path if nodeConfig.BaseDir != "" && !strings.HasPrefix(nodeConfig.AWSGW.Config, "/") { nodeConfig.AWSGW.Config = fmt.Sprintf("%s/%s", nodeConfig.BaseDir, nodeConfig.AWSGW.Config) diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index e143d7a23..58cbd3bf7 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -38,6 +38,7 @@ type Config struct { PluginPath string Debug bool NatsHost string + NatsToken string MountedVolumes []MountedVolume S3Host string Bucket string @@ -143,15 +144,13 @@ func (svc *Service) Reload() (err error) { func launchService(cfg *Config) (err error) { // Connect to NATS - nc, err := nats.Connect(cfg.NatsHost) - + nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken) if err != nil { - slog.Error("Failed to connect to NATS:", "err", err) + slog.Error("Failed to connect to NATS", "err", err) return err } - // Subscribe to the viperblock.mount subject - slog.Info("Connected. Waiting for EBS events") + slog.Info("Waiting for EBS events") if _, err := nc.QueueSubscribe("ebs.delete", "hive-workers", func(msg *nats.Msg) { slog.Info("Received ebs.delete message", "data", string(msg.Data)) diff --git a/hive/utils/nats.go b/hive/utils/nats.go index f188353d0..8df732b60 100644 --- a/hive/utils/nats.go +++ b/hive/utils/nats.go @@ -4,11 +4,39 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "time" "github.com/nats-io/nats.go" ) +// ConnectNATS establishes a connection to a NATS server with standard reconnect +// handling and logging. If token is non-empty, token authentication is used. +func ConnectNATS(host, token string) (*nats.Conn, error) { + opts := []nats.Option{ + nats.ReconnectWait(time.Second), + nats.MaxReconnects(-1), + nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { + slog.Warn("NATS disconnected", "err", err) + }), + nats.ReconnectHandler(func(nc *nats.Conn) { + slog.Info("NATS reconnected", "url", nc.ConnectedUrl()) + }), + } + + if token != "" { + opts = append(opts, nats.Token(token)) + } + + nc, err := nats.Connect(host, opts...) + if err != nil { + return nil, fmt.Errorf("NATS connect failed: %w", err) + } + + slog.Info("Connected to NATS server", "host", host) + return nc, nil +} + // NATSRequest performs a NATS request-response with JSON marshaling. // It marshals the input, sends to the given subject, validates the response // for error payloads, and unmarshals the successful response into Out.