client update - single mq connection

go.mod

@@ -35,7 +35,10 @@ require (
 	github.com/posthog/posthog-go v0.0.0-20211028072449-93c17c49e2b0
 )
 
-require github.com/coreos/go-oidc/v3 v3.2.0
+require (
+	github.com/coreos/go-oidc/v3 v3.2.0
+	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e
+)
 
 require (
 	cloud.google.com/go v0.81.0 // indirect
@@ -61,7 +64,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/goki/freetype v0.0.0-20181231101311-fa8a33aabaff // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/go-cmp v0.5.7 // indirect
+	github.com/google/go-cmp v0.5.8 // indirect
 	github.com/gopherjs/gopherjs v1.17.2 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/josharian/native v1.0.0 // indirect

go.sum

@@ -209,6 +209,7 @@ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
 github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
+github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
@@ -503,6 +504,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
+golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e h1:+WEEuIdZHnUeJJmEUjyYC2gfUMj69yZXw17EnHg/otA=
+golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/image v0.0.0-20220601225756-64ec528b34cd h1:9NbNcTg//wfC5JskFW4Z3sqwVnjmJKHxLAol1bW2qgw=

netclient/functions/daemon.go

@@ -34,6 +34,8 @@ var messageCache = new(sync.Map)
 
 var serverSet map[string]bool
 
+var mqclient mqtt.Client
+
 const lastNodeUpdate = "lnu"
 const lastPeerUpdate = "lpu"
 
@@ -188,12 +190,12 @@ func unsubscribeNode(client mqtt.Client, nodeCfg *config.ClientConfig) {
 func messageQueue(ctx context.Context, wg *sync.WaitGroup, cfg *config.ClientConfig) {
 	defer wg.Done()
 	logger.Log(0, "network:", cfg.Node.Network, "netclient message queue started for server:", cfg.Server.Server)
-	client, err := setupMQTT(cfg, false)
+	err := setupMQTT(cfg, false)
 	if err != nil {
 		logger.Log(0, "unable to connect to broker", cfg.Server.Server, err.Error())
 		return
 	}
-	defer client.Disconnect(250)
+	defer mqclient.Disconnect(250)
 	<-ctx.Done()
 	logger.Log(0, "shutting down message queue for server", cfg.Server.Server)
 }
@@ -228,7 +230,7 @@ func NewTLSConfig(server string) (*tls.Config, error) {
 
 // setupMQTT creates a connection to broker and returns client
 // this function is primarily used to create a connection to publish to the broker
-func setupMQTT(cfg *config.ClientConfig, publish bool) (mqtt.Client, error) {
+func setupMQTT(cfg *config.ClientConfig, publish bool) error {
 	opts := mqtt.NewClientOptions()
 	server := cfg.Server.Server
 	port := cfg.Server.MQPort
@@ -236,7 +238,7 @@ func setupMQTT(cfg *config.ClientConfig, publish bool) (mqtt.Client, error) {
 	tlsConfig, err := NewTLSConfig(server)
 	if err != nil {
 		logger.Log(0, "failed to get TLS config for", server, err.Error())
-		return nil, err
+		return err
 	}
 	opts.SetTLSConfig(tlsConfig)
 	opts.SetClientID(ncutils.MakeRandomString(23))
@@ -286,11 +288,11 @@ func setupMQTT(cfg *config.ClientConfig, publish bool) (mqtt.Client, error) {
 		reRegisterWithServer(cfg)
 		//try after re-registering
 		if token := client.Connect(); !token.WaitTimeout(30*time.Second) || token.Error() != nil {
-			return client, errors.New("unable to connect to broker")
+			return errors.New("unable to connect to broker")
 		}
 	}
-
-	return client, nil
+	mqclient = client
+	return nil
 }
 
 func reRegisterWithServer(cfg *config.ClientConfig) {

netclient/functions/join.go

@@ -210,10 +210,6 @@ func JoinNetwork(cfg *config.ClientConfig, privateKey string) error {
 	if cfg.Server.Server == "" {
 		return errors.New("did not receive broker address from registration")
 	}
-	// update server with latest data
-	if err := PublishNodeUpdate(cfg); err != nil {
-		logger.Log(0, "network:", cfg.Network, "failed to publish update for join", err.Error())
-	}
 
 	if cfg.Daemon == "install" || ncutils.IsFreeBSD() {
 		err = daemon.InstallDaemon()

netclient/functions/mqpublish.go

@@ -21,7 +21,8 @@ import (
 )
 
 // Checkin  -- go routine that checks for public or local ip changes, publishes changes
-//   if there are no updates, simply "pings" the server as a checkin
+//
+//	if there are no updates, simply "pings" the server as a checkin
 func Checkin(ctx context.Context, wg *sync.WaitGroup) {
 	logger.Log(2, "starting checkin goroutine")
 	defer wg.Done()
@@ -141,17 +142,12 @@ func publish(nodeCfg *config.ClientConfig, dest string, msg []byte, qos byte) er
 		return err
 	}
 
-	client, err := setupMQTT(nodeCfg, true)
-	if err != nil {
-		return fmt.Errorf("mq setup error %w", err)
-	}
-	defer client.Disconnect(250)
 	encrypted, err := ncutils.Chunk(msg, serverPubKey, trafficPrivKey)
 	if err != nil {
 		return err
 	}
 
-	if token := client.Publish(dest, qos, false, encrypted); !token.WaitTimeout(30*time.Second) || token.Error() != nil {
+	if token := mqclient.Publish(dest, qos, false, encrypted); !token.WaitTimeout(30*time.Second) || token.Error() != nil {
 		logger.Log(0, "could not connect to broker at "+nodeCfg.Server.Server+":"+nodeCfg.Server.MQPort)
 		var err error
 		if token.Error() == nil {