Cold_handle/MqttServer/MqttClients.go

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)

}