sparrow/services/mqttaccess/access.go

package main

import (
	MQTT "github.com/eclipse/paho.mqtt.golang"
	"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)
}