SubDevMsg

services/emqx-agent/client/mqtt_client.go

@@ -65,6 +65,9 @@ func initMqttClient(conf *MqttConfig) (mc MQTT.Client, err error) {
 		server.Log.Infof("connect to %s", broker.String())
 		return tlsCfg
 	})
+	opts.SetConnectionLostHandler(func(client MQTT.Client, reason error) {
+		server.Log.Errorf("connection lost: %s", reason.Error())
+	})
 	opts.SetOnConnectHandler(func(client MQTT.Client) {
 		if mqttSetOnConnectHandler != nil {
 			mqttSetOnConnectHandler(client)

services/emqx-agent/sub_dev.go

@@ -87,17 +87,17 @@ func (d *MqttClient) SubDevMsg(handle Handle) error {
 	if err != nil {
 		return err
 	}
-	//client.SetMqttSetOnConnectHandler(func(cli mqtt.Client) {
-	//	err := d.subDevMsg(cli, handle)
-	//	if err != nil {
-	//		server.Log.Errorf("mqttSetOnConnectHandler.subDevMsg err:%v", err)
-	//	}
-	//})
+	client.SetMqttSetOnConnectHandler(func(cli mqtt.Client) {
+		err := d.subDevMsg(cli, handle)
+		if err != nil {
+			server.Log.Errorf("mqttSetOnConnectHandler.subDevMsg err:%v", err)
+		}
+	})
 	return nil
 }
 
 func (d *MqttClient) subDevMsg(cli mqtt.Client, handle Handle) error {
-	err := d.subscribeWithFunc(cli, TopicConnectStatus, nil)
+	err := d.subscribeWithFunc(cli, TopicConnectStatus, d.subConnectStatus(handle))
 	server.Log.Infof("subDevMsg")
 	if err != nil {
 		server.Log.Infof("subDevMsg err:%v", err)