SubDevMsg

services/emqx-agent/client/mqtt_client.go

@@ -56,13 +56,19 @@ func initMqttClient(conf *MqttConfig) (mc MQTT.Client, err error) {
 		opts.AddBroker(broker)
 	}
 	opts.SetClientID(fmt.Sprintf("%s_%s", conf.ClientId, guid.S())).
-		SetUsername(conf.User).SetPassword(conf.Password).
-		SetConnectRetry(true).SetCleanSession(true)
+		SetUsername(conf.User).
+		SetPassword(conf.Password).
+		SetConnectRetry(true).
+		SetCleanSession(true).
+		SetKeepAlive(120)
 	opts.SetConnectionAttemptHandler(func(broker *url.URL, tlsCfg *tls.Config) *tls.Config {
 		server.Log.Infof("connect to %s", broker.String())
 		return tlsCfg
 	})
 	opts.SetOnConnectHandler(func(client MQTT.Client) {
+		if mqttSetOnConnectHandler != nil {
+			mqttSetOnConnectHandler(client)
+		}
 		server.Log.Infof("connected success")
 	})
 

services/emqx-agent/sub_dev.go

@@ -83,10 +83,10 @@ func newEmqClient(conf *client.MqttConfig) (SubDev, error) {
 
 func (d *MqttClient) SubDevMsg(handle Handle) error {
 
-	err := d.subDevMsg(nil, handle)
-	if err != nil {
-		return err
-	}
+	//err := d.subDevMsg(nil, handle)
+	//if err != nil {
+	//	return err
+	//}
 	client.SetMqttSetOnConnectHandler(func(cli mqtt.Client) {
 		err := d.subDevMsg(cli, handle)
 		if err != nil {