update

pkg/mqtt/connection.go

@@ -287,6 +287,8 @@ func (c *Connection) RcvMsgFromClient() {
 
 		case *Subscribe:
 			server.Log.Infof("%s, subscribe topic: %v", c.DeviceCode, msg.Topics)
+			subAck := SubAck{MessageID: msg.MessageID}
+			c.Submit(&subAck)
 
 		case *Unsubscribe:
 			server.Log.Infof("%s, unsubscribe topic: %v", host, msg.Topics)

services/mqttaccess/main.go

@@ -46,17 +46,9 @@ func main() {
 		Description: "MQTT Device Online Count",
 		Type:        "gauge",
 	}
-	recvMessageCount := &server.Metric{
-		ID:          "device_message_total",
-		Name:        "device_message_total",
-		Description: "device report message total count",
-		Type:        "counter_vec",
-		Args:        []string{"device_code"},
-	}
 
-	server.RegisterMetric(deviceOnlineMetric, recvMessageCount)
+	server.RegisterMetric(deviceOnlineMetric)
 	deviceOnlineCount = deviceOnlineMetric.MetricCollector.(prometheus.Gauge)
-	deviceMessageCount = recvMessageCount.MetricCollector.(*prometheus.CounterVec)
 
 	go func() {
 		pprofPort := "6060"

services/mqttaccess/mqtt_provider.go

@@ -81,7 +81,6 @@ func (mp *MQTTProvider) OnDeviceOffline(deviceid string, vendorId string) error
 	return err
 }
 func (mp *MQTTProvider) OnDeviceHeartBeat(deviceid string) error {
-	deviceMessageCount.WithLabelValues(deviceid).Inc()
 	args := rpcs.ArgsDeviceId{
 		Id: deviceid,
 	}
@@ -93,7 +92,6 @@ func (mp *MQTTProvider) OnDeviceHeartBeat(deviceid string) error {
 	return err
 }
 func (mp *MQTTProvider) OnDeviceMessage(deviceid, vendorId string, msgtype string, message *gjson.Json) {
-	deviceMessageCount.WithLabelValues(deviceid).Inc()
 	server.Log.Debugf("device {%v} message {%v} : %s", deviceid, msgtype, message.MustToJsonString())
 	switch msgtype {
 	case "s":