update //过滤诊断数据 wjh需求

adaptors/安心云最新设备数据toES.go

@@ -25,6 +25,14 @@ func (the Adaptor_AXY_LastRAW) Transform(topic, rawMsg string) *models.EsRaw {
 		log.Printf("反序列化 异常 dev数据=%s", rawMsg)
 		return nil
 	}
+
+	//过滤诊断数据
+	if v, ok := iotaData.Data.Data["_data_type"]; ok {
+		if v == "diag" {
+			return nil
+		}
+	}
+
 	if iotaData.DeviceId == the.TagDeviceId {
 		log.Printf("--> Transform 标记设备数据 [%s] %s ", iotaData.DeviceId, rawMsg)
 	}

consumers/consumerAXYraw.go

@@ -161,8 +161,8 @@ func (the *consumerAXYraw) onData(topic string, msg string) bool {
 		}
 		the.dataCache <- needPush
 	} else {
-		s, _ := json.Marshal(needPush)
 		if needPush != nil {
+			s, _ := json.Marshal(needPush)
 			if needPush.IotaDevice == logTagDeviceId {
 				log.Printf("onData 设备[%s] 异常needPush=  %s", needPush.IotaDevice, s)
 			}