sparrow/services/mqttaccess/access.go

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
}