瀏覽代碼

mq test for scalability

Abhishek Kondur 2 年之前
父節點
當前提交
ddbe5790a5
共有 1 個文件被更改,包括 6 次插入2 次删除
  1. 6 2
      mq/mq.go

+ 6 - 2
mq/mq.go

@@ -36,6 +36,7 @@ func setMqOptions(user, password string, opts *mqtt.ClientOptions) {
 	opts.SetConnectRetryInterval(time.Second << 2)
 	opts.SetKeepAlive(time.Minute)
 	opts.SetWriteTimeout(time.Minute)
+	opts.SetCleanSession(true)
 }
 
 // SetupMQTT creates a connection to broker and return client
@@ -80,8 +81,11 @@ func SetupMQTT() {
 			logger.Log(0, "node metrics subscription failed")
 		}
 
-		opts.SetOrderMatters(true)
-		opts.SetResumeSubs(true)
+		opts.SetOrderMatters(false)
+		opts.SetResumeSubs(false)
+	})
+	opts.SetConnectionLostHandler(func(c mqtt.Client, e error) {
+		setMqOptions(servercfg.GetMqUserName(), servercfg.GetMqPassword(), opts)
 	})
 	mqclient = mqtt.NewClient(opts)
 	tperiod := time.Now().Add(10 * time.Second)