package main import ( "context" "github.com/gogf/gf/encoding/gjson" "sparrow/pkg/klink" "sparrow/pkg/models" "sparrow/pkg/rpcs" "sparrow/pkg/server" "time" ) type MQTTProvider struct{} func NewMQTTProvider() *MQTTProvider { return &MQTTProvider{} } func (mp *MQTTProvider) ValidateDeviceToken(deviceid string, token []byte) error { args := rpcs.ArgsValidateDeviceAccessToken{ Id: deviceid, AccessToken: token, } reply := rpcs.ReplyValidateDeviceAccessToken{} err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.ValidateDeviceAccessToken", args, &reply) if err != nil { return err } return nil } func (mp *MQTTProvider) OnDeviceOnline(args rpcs.ArgsGetOnline, VendorId string) error { deviceOnlineCount.Inc() reply := rpcs.ReplyGetOnline{} err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetOnline", args, &reply) if err != nil { server.Log.Errorf("device online error. args: %v, error: %v", args, err) } var cReply rpcs.ReplyEmptyResult var cArgs rpcs.ArgsGetStatus cArgs.VendorId = VendorId cArgs.Id = args.Id if err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.Online", &cArgs, &cReply); err != nil { return err } // 下发时间同步指令 var cmdArgs rpcs.ArgsSendCommand cmdArgs.Cmd = "timeSync" cmdArgs.DeviceId = args.Id cmdArgs.SubDevice = "01" cmdArgs.Params = map[string]interface{}{ "value": time.Now().Unix(), } err = server.RPCCallByName(nil, rpcs.MQTTAccessName, "Access.SendCommand", cmdArgs, &cReply) if err != nil { server.Log.Errorf("发送时间同步指令:%v", cmdArgs, err) } return err } func (mp *MQTTProvider) OnDeviceOffline(deviceid string, vendorId string) error { if deviceid != "" { deviceOnlineCount.Dec() } args := rpcs.ArgsGetOffline{ Id: deviceid, VendorId: vendorId, } reply := rpcs.ReplyGetOffline{} err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetOffline", args, &reply) if err != nil { server.Log.Errorf("device offline error. deviceid: %v, error: %v", deviceid, err) } var cReply rpcs.ReplyEmptyResult var cArgs rpcs.ArgsGetStatus cArgs.VendorId = vendorId cArgs.Id = args.Id if err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.Offline", &cArgs, &cReply); err != nil { return err } return err } func (mp *MQTTProvider) OnDeviceHeartBeat(deviceid string) error { deviceMessageCount.WithLabelValues(deviceid).Inc() args := rpcs.ArgsDeviceId{ Id: deviceid, } reply := rpcs.ReplyHeartBeat{} err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.HeartBeat", args, &reply) if err != nil { server.Log.Errorf("device heartbeat error. deviceid: %v, error: %v", deviceid, err) } return err } func (mp *MQTTProvider) OnDeviceMessage(deviceid, vendorId string, msgtype string, message *gjson.Json) { deviceMessageCount.WithLabelValues(deviceid).Inc() server.Log.Infof("device {%v} message {%v} : %s", deviceid, msgtype, message.MustToJsonString()) switch msgtype { case "s": act := klink.PacketAction(message.GetString("action")) if act != "" { switch act { case klink.DevSendAction: processReportStatus(deviceid, vendorId, message) case klink.DevLoginAction: _ = processDevLogin(deviceid, message.GetString("subDeviceId")) case klink.DevLogoutAction: _ = processDevLogout(deviceid, message.GetString("subDeviceId")) case klink.DevNetConfigAction: _ = processDevNetConfig(deviceid, message.GetString("md5")) case klink.ReportFirmwareAction: _ = processDeviceReportUpgrade(deviceid, message.GetString("version")) } } case "e": reply := rpcs.ReplyOnEvent{} args := rpcs.ArgsOnEvent{ DeviceId: deviceid, TimeStamp: message.GetUint64("timestamp"), SubDeviceId: message.GetString("subDeviceId"), SubData: message.GetJson("data").MustToJson(), VendorId: vendorId, } err := server.RPCCallByName(nil, rpcs.ControllerName, "Controller.OnEvent", args, &reply) if err != nil { server.Log.Errorf("device on event error. args: %v, error: %v", args, err) return } default: server.Log.Infof("unknown message type: %v", msgtype) } } func processReportStatus(deviceid, vendorId string, message *gjson.Json) { reply := rpcs.ReplyOnStatus{} args := rpcs.ArgsOnStatus{ DeviceId: deviceid, Timestamp: message.GetUint64("timestamp"), SubData: message.GetJson("data").MustToJson(), VendorId: vendorId, SubDeviceId: message.GetString("subDeviceId"), Action: klink.PacketAction(message.GetString("action")), } err := server.RPCCallByName(nil, rpcs.ControllerName, "Controller.OnStatus", args, &reply) if err != nil { server.Log.Errorf("device report status error. args: %v, error: %v", args, err) return } } func processDevLogin(deviceCode, subDeviceId string) error { var args rpcs.SubDeviceArgs args.DeviceCode = deviceCode args.Status = 1 args.SubDeviceId = subDeviceId var reply *models.SubDevice err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.UpdateSubDevice", &args, &reply) if err != nil { server.Log.Errorf("子设备上线出错:%s", err.Error()) } return nil } func processDevLogout(deviceCode, subDeviceId string) error { var args rpcs.SubDeviceArgs args.DeviceCode = deviceCode args.Status = 0 args.SubDeviceId = subDeviceId var reply *models.SubDevice err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.UpdateSubDevice", &args, &reply) if err != nil { server.Log.Errorf("子设备下线出错:%s", err.Error()) } return nil } // 处理设备配网信息 func processDevNetConfig(deviceCode, md5 string) error { args := &models.DeviceNetConfig{ DeviceIdentifier: deviceCode, MD5: md5, Status: 1, } reply := rpcs.ReplyCheckDeviceNetConfig{} err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.CreateDeviceNetConfig", args, &reply) if err != nil { server.Log.Errorf("set device:%s net config info error:%v", deviceCode, err) } return nil } // 设备上报固件信息处理 func processDeviceReportUpgrade(deviceId, version string) error { args := &rpcs.ArgsUpdateDeviceVersion{ DeviceId: deviceId, Version: version, } var reply rpcs.ReplyEmptyResult err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.UpdateDeviceVersion", args, &reply) if err != nil { server.Log.Errorf("更新设备版本号失败:%v", args, err) } return nil }