123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195 |
- 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
- }
|