zishuo/netmaker
1
0
Fork 0
mirror of https://github.com/gravitl/netmaker.git synced 2025-10-04 16:33:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

added new struct and adapted to funcs

Browse Source
This commit is contained in:
0xdcarns
2022-01-26 10:40:39 -05:00
parent 0d3813295d
commit cd4e2c57d6
7 changed files with 42 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
netclient/functions/daemon.go
View File

@@ -8,7 +8,6 @@ import (
"os"
"os/signal"
"runtime"
"strings"
"sync"
"syscall"
"time"
@@ -60,11 +59,10 @@ func Daemon() error {
// SetupMQTT creates a connection to broker and return client
func SetupMQTT(cfg *config.ClientConfig) mqtt.Client {
opts := mqtt.NewClientOptions()
serverAddrs := strings.Split(cfg.Node.NetworkSettings.DefaultServerAddrs, ",")
for i, addr := range serverAddrs {
if addr != "" {
for i, server := range cfg.Node.NetworkSettings.DefaultServerAddrs {
if server.Address != "" && server.IsLeader {
ncutils.Log(fmt.Sprintf("adding server (%d) to listen on network %s \n", (i + 1), cfg.Node.Network))
opts.AddBroker(addr + ":1883")
opts.AddBroker(server.Address + ":1883")
}
}
opts.SetDefaultPublishHandler(All)
@@ -89,13 +87,13 @@ func MessageQueue(ctx context.Context, network string) {
}
ncutils.Log("subscribed to all topics for debugging purposes")
}
if token := client.Subscribe("update/"+cfg.Node.ID, 0, NodeUpdate); token.Wait() && token.Error() != nil {
if token := client.Subscribe("update/"+cfg.Node.ID, 0, mqtt.MessageHandler(NodeUpdate)); token.Wait() && token.Error() != nil {
log.Fatal(token.Error())
}
if cfg.DebugOn {
ncutils.Log("subscribed to node updates for node " + cfg.Node.Name + " update/" + cfg.Node.ID)
}
if token := client.Subscribe("update/peers/"+cfg.Node.ID, 0, UpdatePeers); token.Wait() && token.Error() != nil {
if token := client.Subscribe("update/peers/"+cfg.Node.ID, 0, mqtt.MessageHandler(UpdatePeers)); token.Wait() && token.Error() != nil {
log.Fatal(token.Error())
}
if cfg.DebugOn {
@@ -115,7 +113,7 @@ var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
}
// NodeUpdate -- mqtt message handler for /update/<NodeID> topic
var NodeUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
ncutils.Log("received message to update node " + string(msg.Payload()))
//potentiall blocking i/o so do this in a go routine
go func() {
@@ -199,7 +197,7 @@ var NodeUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
}
// UpdatePeers -- mqtt message handler for /update/peers/<NodeID> topic
var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
func UpdatePeers(client mqtt.Client, msg mqtt.Message) {
go func() {
var peerUpdate models.PeerUpdate
err := json.Unmarshal(msg.Payload(), &peerUpdate)
@@ -217,12 +215,10 @@ var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
var cfg config.ClientConfig
cfg.Network = peerUpdate.Network
cfg.ReadConfig()
var shouldReSub = shouldResub(strings.Split(cfg.Node.NetworkSettings.DefaultServerAddrs, ","), peerUpdate.ServerAddrs)
var shouldReSub = shouldResub(cfg.Node.NetworkSettings.DefaultServerAddrs, peerUpdate.ServerAddrs)
if shouldReSub {
client.Disconnect(250) // kill client
// un sub, re sub.. how?
client.Unsubscribe("update/"+cfg.Node.ID, "update/peers/"+cfg.Node.ID)
cfg.Node.NetworkSettings.DefaultServerAddrs = strings.Join(peerUpdate.ServerAddrs, ",")
Resubscribe(client, &cfg)
cfg.Node.NetworkSettings.DefaultServerAddrs = peerUpdate.ServerAddrs
}
file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
err = wireguard.UpdateWgPeers(file, peerUpdate.Peers)
@@ -240,8 +236,9 @@ var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
}
// Resubscribe --- handles resubscribing if needed
func Resubscribe(client mqtt.Client, cfg *config.ClientConfig) {
func Resubscribe(client mqtt.Client, cfg *config.ClientConfig) error {
if err := config.ModConfig(&cfg.Node); err == nil {
ncutils.Log("resubbing on network " + cfg.Node.Network)
client.Disconnect(250)
client = SetupMQTT(cfg)
if token := client.Subscribe("update/"+cfg.Node.ID, 0, NodeUpdate); token.Wait() && token.Error() != nil {
@@ -254,8 +251,10 @@ func Resubscribe(client mqtt.Client, cfg *config.ClientConfig) {
log.Fatal(token.Error())
}
ncutils.Log("finished re subbing")
return nil
} else {
ncutils.Log("could not mod config when re-subbing")
return err
}
}
@@ -355,12 +354,12 @@ func Hello(cfg *config.ClientConfig, network string) {
client.Disconnect(250)
}
func shouldResub(currentServers, newServers []string) bool {
func shouldResub(currentServers, newServers []models.ServerAddr) bool {
if len(currentServers) != len(newServers) {
return false
return true
}
for _, srv := range currentServers {
if !ncutils.StringSliceContains(newServers, srv) {
if !ncutils.ServerAddrSliceContains(newServers, srv) {
return true
}
}