package main import ( "github.com/gogf/gf/encoding/gjson" "sparrow/pkg/klink" "sparrow/pkg/models" "sparrow/pkg/rpcs" "sparrow/pkg/server" ) 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 { server.Log.Errorf("validate device token error. deviceid : %v, token : %v, error: %v", deviceid, token, err) 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 } 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 "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 { server.Log.Debugf("子设备上线:%s,%s", deviceCode, subDeviceId) return nil } func processDevLogout(deviceCode, subDeviceId string) error { server.Log.Debugf("子设备下线:%s,%s", deviceCode, subDeviceId) 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 }