package main import ( "bytes" "encoding/json" "errors" "fmt" MQTT "github.com/eclipse/paho.mqtt.golang" "github.com/gogf/gf/v2/encoding/gbinary" "github.com/gogf/gf/v2/util/gconv" "sparrow/pkg/klink" "sparrow/pkg/mqtt" "sparrow/pkg/protocol" "sparrow/pkg/rpcs" "sparrow/pkg/server" "time" ) const ( defaultTimeoutSecond = 5 commandGetCurrentStatus = uint16(65528) ) type Access struct { MqttBroker *mqtt.Broker MQTT.Client } func NewAccess() (*Access, error) { p := NewMQTTProvider() return &Access{ MqttBroker: mqtt.NewBroker(p), }, nil } func (a *Access) SetStatus(args rpcs.ArgsSetStatus, reply *rpcs.ReplySetStatus) error { server.Log.Infof("Access Set Status: %v", args) data := &protocol.Data{} data.Head.Timestamp = uint64(time.Now().Unix()) token, err := a.MqttBroker.GetToken(args.DeviceId) if err != nil { return err } copy(data.Head.Token[:], token[:16]) data.SubData = args.Status msg, err := data.Marshal() if err != nil { return err } return a.MqttBroker.SendMessageToDevice(args.DeviceId, "s", msg, defaultTimeoutSecond*time.Second) } func (a *Access) 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 (a *Access) SendCommand(args rpcs.ArgsSendCommand, reply *rpcs.ReplySendCommand) error { 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 } print("Access Send Command: %v, %v,%s\r\n", args.DeviceId, args.Cmd, string(msg)) return a.MqttBroker.SendMessageToDevice(args.DeviceId, "c", msg, time.Duration(args.WaitTime)*time.Second) } // Upgrade 设备OTA升级指令 func (a *Access) Upgrade(args rpcs.ArgsDeviceUpgrade, reply *rpcs.ReplyEmptyResult) error { server.Log.Infof("设备OTA升级:%s, %s", args.DeviceId, args.Version) cmd := &klink.CloudSend{ Action: "cloudSend", MsgId: 0, DeviceCode: args.DeviceId, Timestamp: time.Now().Unix(), Data: &klink.CloudSendData{ Cmd: "devUpgrade", Params: map[string]interface{}{ "md5": args.Md5, "url": args.Url, "version": args.Version, "file_size": args.FileSize, }, }, SubDeviceId: args.SudDeviceId, } msg, err := cmd.Marshal() if err != nil { return err } return a.MqttBroker.SendMessageToDevice(args.DeviceId, "c", msg, 5*time.Second) } // UpgradeInfo 下发升级包信息 // TODO: 实现 func (a *Access) UpgradeFor4G(args rpcs.ArgsUpgrade4G, reply *rpcs.ReplyEmptyResult) error { server.Log.Infof("4G模组OTA升级:%s", args.DeviceId) cmd := &klink.CloudSend{ Action: "cloudSend", MsgId: 0, DeviceCode: args.DeviceId, Timestamp: time.Now().Unix(), Data: &klink.CloudSendData{ Cmd: "devUpgrade", Params: map[string]interface{}{ "fileId": args.FileId, "fileSize": args.FileSize, }, }, } msg, err := cmd.Marshal() if err != nil { return err } return a.MqttBroker.SendMessageToDevice(args.DeviceId, "c", msg, 5*time.Second) } func (a *Access) ChunkUpgrade(args rpcs.ChunkUpgrade, reply *rpcs.ReplyEmptyResult) error { server.Log.Infof("4G模组OTA升级:%s", args.DeviceId) cmd := &klink.CloudSend{ Action: "cloudSend", MsgId: 0, DeviceCode: args.DeviceId, Timestamp: time.Now().Unix(), Data: &klink.CloudSendData{ Cmd: "devUpgrade", Params: map[string]interface{}{ "fileId": args.FileId, "fileSize": args.FileSize, "size": args.Size, "offset": args.Offset, }, }, } byteCmd, err := json.Marshal(cmd) if err != nil { return err } buf := bytes.NewBuffer(gbinary.BeEncodeUint16(gconv.Uint16(len(byteCmd)))) buf.Write(byteCmd) var fileArgs rpcs.ArgsOtaFile fileArgs.FileId = args.FileId var fileReply rpcs.ReplyOtaFile err = server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetFile", fileArgs, &reply) if err != nil { server.Log.Errorf("OTA升级文件保存失败:%v", err) return err } if fileReply.File == nil { return errors.New(fmt.Sprintf("文件:%s 获取失败", args.FileId)) } buf.Write(fileReply.File[args.Offset : args.Offset+int(args.Size)]) var mCrc crc checkSum := mCrc.reset().pushBytes(buf.Bytes()).value() buf.Write([]byte{byte(checkSum), byte(checkSum >> 8)}) var SendByteArgs rpcs.ArgsSendByteData SendByteArgs.DeviceId = args.DeviceId SendByteArgs.Data = buf.Bytes() err = server.RPCCallByName(nil, rpcs.EmqxAgentServiceName, "Access.SendByteData", SendByteArgs, &reply) return nil }