package main import ( "context" "github.com/gogf/gf/encoding/gjson" "sparrow/pkg/klink" "sparrow/pkg/models" "sparrow/pkg/protocol" "sparrow/pkg/rpcs" "sparrow/pkg/server" ) type Agent struct { } // Message 收到设备上报消息处理 func (a *Agent) Message(topic string, payload []byte) error { topicInfo, err := protocol.GetTopicInfo(topic) if err != nil { return err } if topicInfo.Direction == protocol.Down { return nil } device := &models.Device{} err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", topicInfo.DeviceCode, device) if err != nil { server.Log.Errorf("device not found %s", topicInfo.DeviceCode) return nil } server.Log.Debugf("device {%v} message {%v} : %s", device.DeviceIdentifier, topicInfo.Types, payload) if len(topicInfo.Types) == 0 { return nil } jsonPayload, err := gjson.DecodeToJson(payload) if err != nil { return nil } switch topicInfo.Types[0] { case "status": return a.processStatus(topicInfo.DeviceCode, device.VendorID, jsonPayload) case "event": return a.processEvent(topicInfo.DeviceCode, device.VendorID, jsonPayload) } return nil } func (a *Agent) processEvent(deviceId, vendorId string, message *gjson.Json) error { 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 err } return nil } func (a *Agent) processStatus(deviceId, vendorId string, message *gjson.Json) error { 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")) } } return nil } 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", err) } return nil } 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 } } // Connected 设备接入时 func (a *Agent) Connected(status *protocol.DevConnectStatus) error { // 查询设备信息 device := &models.Device{} err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", status.DeviceId, device) if err != nil { server.Log.Errorf("device not found %s", status.DeviceId) return nil } args := rpcs.ArgsGetOnline{ Id: device.RecordId, ClientIP: status.ClientIp, AccessRPCHost: server.GetRPCHost(), HeartbeatInterval: 300, } reply := rpcs.ReplyGetOnline{} err = server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetOnlineV2", 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 = device.VendorID cArgs.Id = args.Id if err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.Online", &cArgs, &cReply); err != nil { return err } return nil } // Disconnected 设备断开连接时 func (a *Agent) Disconnected(status *protocol.DevConnectStatus) error { // 查询设备信息 device := &models.Device{} err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", status.DeviceId, device) if err != nil { server.Log.Errorf("device not found %s", status.DeviceId) return nil } args := rpcs.ArgsGetOffline{ Id: device.RecordId, VendorId: device.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", status.DeviceId, err) } return err } func NewAgent() *Agent { return &Agent{} }