package MqttServer import ( "Cold_mqtt/MqttServer/WarningNotice" "Cold_mqtt/conf" "Cold_mqtt/lib" "Cold_mqtt/logs" "Cold_mqtt/models/Device" "Cold_mqtt/models/Warning" "encoding/json" "strings" "time" ) // 设备连接 func MessageConnected(topicName string, message []byte) { logs.Println("============= MessageConnected Mqtt JSON =============") logs.Println("topic:", topicName, " message:", string(message)) //logs.Println("=>", a) var Ms_project Ms_Project_Connected err := json.Unmarshal(message, &Ms_project) if err != nil { logs.PrintlnError("MqttServer", "JSON反序列化失败[Ms_Project]", string(message)) return } Clientid_list := strings.Split(Ms_project.Clientid+"_", "_") Ms_project.Clientid = Clientid_list[0] // 过滤 if len(Ms_project.Clientid) < 5 { return } r_Device, err := Device.Read_Device_ByT_sn(Ms_project.Clientid) if err != nil { logs.Println("MessageDisconnected 没有该设备:", Ms_project.Clientid) return } logs.Println("Clientid:", Ms_project.Clientid) logs.Println("Username:", Ms_project.Username) Reason := "" if strings.Contains(Ms_project.Username, "_s") { r_Device.T_online_s = 1 Reason = "备用网络|IP:" + Ms_project.Ipaddress } else { r_Device.T_online = 1 Reason = "主网络|IP:" + Ms_project.Ipaddress } Warning.Add_DeviceLogs(103, r_Device, Reason) r_Device.T_mqttid = conf.MqttServer_id // 同步参数 Device.Update_Device(r_Device, "T_mqttid", "T_online", "T_online_s") Device.Update_Device_To_DeviceSensor(r_Device) // 设备同步参数 // 监查 设备参数 go func() { time.Sleep(time.Second * 10) logs.Println("上线检测 待设置 设备参数:", r_Device.T_sn) DeviceParameter_list := Device.Read_DeviceParameter_SN_T_SendState_0(r_Device.T_sn) if len(DeviceParameter_list) > 0 { logs.Println("Pu_DeviceParameter->:", DeviceParameter_list[0].Id, DeviceParameter_list[0].T_sn) Pu_DeviceParameter(DeviceParameter_list[0]) } logs.Println("上线检测 待设置 传感器参数:", r_Device.T_sn) DeviceSensorParameter_list := Device.Read_DeviceSensorParameter_SN_T_SendState_0_sql(r_Device.T_sn) for i, v := range DeviceSensorParameter_list { sn := v[0].(string) t_id := v[1].(string) logs.Println("DeviceSensorParameter_list:", i, sn, t_id) r_DeviceSensorParameter_list := Device.Read_DeviceSensorParameter_SN_T_id(sn, lib.To_int(t_id)) if len(r_DeviceSensorParameter_list) > 0 { if r_DeviceSensorParameter_list[0].T_SendState == 0 { logs.Println("Pu_DeviceParameter_Sensor->:", i, sn, t_id) Pu_DeviceParameter_Sensor(r_DeviceSensorParameter_list[0]) } } } }() logs.Println("============= MessageConnected Mqtt JSON AND =============", topicName) } func ToReason(str string) (Reason string) { switch str { case "normal": Reason = "客户端主动断开" break case "kicked": Reason = "设备异常掉线,服务端踢出,通过 REST API" break case "keepalive_timeout": Reason = "设备异常掉线,keepalive 超时" break case "not_authorized": Reason = "设备异常掉线,认证失败" break case "tcp_closed": Reason = "设备异常掉线,对端关闭了网络连接" break case "internal_error": Reason = "设备异常掉线,畸形报文或其他未知错误" break } return Reason } // 设备断开 func MessageDisconnected(topicName string, message []byte) { logs.Println("============= MessageDisconnected Mqtt JSON =============") logs.Println("topic:", topicName, " message:", string(message)) //logs.Println("=>", a) var Ms_project Ms_Project_Disconnected err := json.Unmarshal(message, &Ms_project) if err != nil { logs.PrintlnError("MqttServer", "JSON反序列化失败[Ms_Project]", string(message)) return } logs.Println("Clientid:", Ms_project.Clientid) logs.Println("Username:", Ms_project.Username) Clientid_list := strings.Split(Ms_project.Clientid+"_s", "_") Ms_project.Clientid = Clientid_list[0] // 过滤 if len(Ms_project.Clientid) < 5 { return } r_Device, err := Device.Read_Device_ByT_sn(Ms_project.Clientid) if err != nil { logs.Println("MessageDisconnected 没有该设备:", Ms_project.Clientid) return } Reason := ToReason(Ms_project.Reason) if strings.Contains(Ms_project.Username, "_s") { r_Device.T_online_s = 2 Reason = "备用网络|" + Reason } else { r_Device.T_online = 2 Reason = "主网络|" + Reason } if !strings.Contains(Reason, "主动断开") && r_Device.T_model == "BX100W" { Reason = "可能市电断电,请注意查看" } Warning_r := Warning.Add_DeviceLogs(1001, r_Device, Reason) Warning_r.T_fUt = time.Now() if !strings.Contains(Reason, "主动断开") { // 异常断开 //WarningNotice.WarningToAdminId(&Warning_r, []int{}) // 发送给 绑定公司管理员 //// 报警处理 //Warning.Add_Warning_Log(&Warning_r, "====== 报警策略 ===== \n") WarningNotice.WarningCompanyNotice(&Warning_r, 0, 0) // 移动端 监控状态同步问题(2小时后自动拉为未监控状态) if strings.Contains(r_Device.T_model, "BW100") { go func(sn string) { time_s := time.Now() time.Sleep(2 * time.Hour) r_Device_, err := Device.Read_Device_ByT_sn(sn) if err != nil { logs.Println("MessageDisconnected 没有该设备:", Ms_project.Clientid) return } if r_Device_.T_online != 1 { r_Device_.T_monitor = 0 // 同步参数 Device.Update_Device(r_Device, "T_monitor") Device.Update_Device_To_DeviceSensor(r_Device) // 设备同步参数 logs.Println(sn + ",在" + time_s.Format("2006-01-02 15:04:05") + "异常下线, 在两个小时后 未能上线,将其监控状态强制拉为未监控!") } else { logs.Println(sn + ",在" + time_s.Format("2006-01-02 15:04:05") + "异常下线, 在两个小时后 成功上线!") } }(r_Device.T_sn) } } else { // 正常关机 r_Device.T_monitor = 0 if !strings.Contains(r_Device.T_model, "BW100") { DeviceSensorList := Device.Read_DeviceSensor_ByTsn(r_Device.T_sn) logs.Println("DeviceSensorList:", DeviceSensorList) for _, v := range DeviceSensorList { DeviceData_t, is := Device.RedisDeviceData_Get(r_Device.T_sn+"_Redis", v.T_id) logs.Println("DeviceData_t is:", is) if is { Device.Add_DeviceData(r_Device.T_sn, v.T_id, DeviceData_t, true) Device.RedisDeviceData_Del(r_Device.T_sn+"_Redis", v.T_id) // 清除缓存标志 } } } } // 同步参数 Device.Update_Device(r_Device, "T_online", "T_online_s", "T_monitor") Device.Update_Device_To_DeviceSensor(r_Device) // 设备同步参数 logs.Println("============= MessageDisconnected Mqtt JSON AND =============", topicName) }