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" "time" ) type Agent struct { client SubDev } // 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.DeviceIdentifier, 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.DeviceIdentifier, 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 } // SendCommand rpc 发送设备命令 func (a *Agent) SendCommand(args rpcs.ArgsSendCommand, reply *rpcs.ReplySendCommand) error { // 查询设备信息 device := &models.Device{} err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", args.DeviceId, device) if err != nil { server.Log.Errorf("device not found %s", args.DeviceId) return nil } product := &models.Product{} err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindProduct", device.ProductID, product) if err != nil { server.Log.Errorf("device not found %s", args.DeviceId) return nil } cmd := &klink.CloudSend{ Action: "cloudSend", MsgId: 0, DeviceCode: args.DeviceId, SubDeviceId: args.SubDevice, Timestamp: time.Now().Unix(), Data: &klink.CloudSendData{ Cmd: args.Cmd, Params: args.Params, }, } msg, err := cmd.Marshal() if err != nil { return err } return a.client.PublishToMsgToDev(protocol.GetCommandTopic(args.DeviceId, product.ProductKey), msg) } // GetStatus rpc 获取设备状态 func (a *Agent) GetStatus(args rpcs.ArgsGetStatus, reply *rpcs.ReplyGetStatus) error { server.Log.Infof("Access Get Status: %v", args) // first send a get status command cmdArgs := rpcs.ArgsSendCommand{ DeviceId: args.Id, WaitTime: 0, SubDevice: args.SubDeviceId, Cmd: "report", } cmdReply := rpcs.ReplySendCommand{} return a.SendCommand(cmdArgs, &cmdReply) } func NewAgent(client SubDev) *Agent { return &Agent{ client: client, } }