netmaker/mq/mq.go

85 lines
2.5 KiB
Go
Raw Normal View History

2022-01-13 05:23:34 +08:00
package mq
import (
"context"
"log"
"time"
2022-01-13 05:23:34 +08:00
mqtt "github.com/eclipse/paho.mqtt.golang"
"github.com/gravitl/netmaker/logger"
2022-02-09 04:05:37 +08:00
"github.com/gravitl/netmaker/netclient/ncutils"
2022-01-19 06:32:12 +08:00
"github.com/gravitl/netmaker/servercfg"
2022-01-13 05:23:34 +08:00
)
2022-02-09 08:13:58 +08:00
// KEEPALIVE_TIMEOUT - time in seconds for timeout
2022-02-02 03:01:13 +08:00
const KEEPALIVE_TIMEOUT = 60 //timeout in seconds
2022-02-09 08:13:58 +08:00
// MQ_DISCONNECT - disconnects MQ
const MQ_DISCONNECT = 250
2022-05-16 21:38:47 +08:00
// MQ_TIMEOUT - timeout for MQ
const MQ_TIMEOUT = 30
2022-02-09 10:49:47 +08:00
var peer_force_send = 0
// SetupMQTT creates a connection to broker and return client
2022-02-09 08:13:58 +08:00
func SetupMQTT(publish bool) mqtt.Client {
2022-01-19 06:32:12 +08:00
opts := mqtt.NewClientOptions()
2022-02-09 08:13:58 +08:00
opts.AddBroker(servercfg.GetMessageQueueEndpoint())
2022-02-09 04:05:37 +08:00
id := ncutils.MakeRandomString(23)
opts.ClientID = id
2022-02-09 08:13:58 +08:00
opts.SetAutoReconnect(true)
opts.SetConnectRetry(true)
opts.SetConnectRetryInterval(time.Second << 2)
opts.SetKeepAlive(time.Minute)
opts.SetWriteTimeout(time.Minute)
opts.SetOnConnectHandler(func(client mqtt.Client) {
if !publish {
2022-05-16 21:38:47 +08:00
if token := client.Subscribe("ping/#", 2, mqtt.MessageHandler(Ping)); token.WaitTimeout(MQ_TIMEOUT*time.Second) && token.Error() != nil {
2022-02-09 08:13:58 +08:00
client.Disconnect(240)
logger.Log(0, "ping subscription failed")
}
2022-05-16 21:38:47 +08:00
if token := client.Subscribe("update/#", 0, mqtt.MessageHandler(UpdateNode)); token.WaitTimeout(MQ_TIMEOUT*time.Second) && token.Error() != nil {
2022-02-09 08:13:58 +08:00
client.Disconnect(240)
logger.Log(0, "node update subscription failed")
}
2022-05-16 21:38:47 +08:00
if token := client.Subscribe("signal/#", 0, mqtt.MessageHandler(ClientPeerUpdate)); token.WaitTimeout(MQ_TIMEOUT*time.Second) && token.Error() != nil {
client.Disconnect(240)
logger.Log(0, "node client subscription failed")
}
2022-02-09 08:13:58 +08:00
opts.SetOrderMatters(true)
opts.SetResumeSubs(true)
}
})
2022-01-28 04:52:31 +08:00
client := mqtt.NewClient(opts)
2022-01-28 04:46:45 +08:00
tperiod := time.Now().Add(10 * time.Second)
for {
2022-05-16 21:38:47 +08:00
if token := client.Connect(); token.WaitTimeout(MQ_TIMEOUT*time.Second) && token.Error() != nil {
2022-01-28 04:46:45 +08:00
logger.Log(2, "unable to connect to broker, retrying ...")
2022-01-28 04:52:31 +08:00
if time.Now().After(tperiod) {
2022-05-16 21:38:47 +08:00
if token.Error() == nil {
log.Fatal(0, "could not connect to broker, token timeout, exiting ...")
} else {
log.Fatal(0, "could not connect to broker, exiting ...", token.Error())
}
2022-01-28 04:46:45 +08:00
}
} else {
break
}
time.Sleep(2 * time.Second)
2022-01-19 06:32:12 +08:00
}
return client
2022-01-19 06:32:12 +08:00
}
// Keepalive -- periodically pings all nodes to let them know server is still alive and doing well
func Keepalive(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case <-time.After(time.Second * KEEPALIVE_TIMEOUT):
2022-02-09 10:34:38 +08:00
sendPeers()
}
}
}