sparrow/services/mqttaccess/access.go

package main

import (
	"github.com/gogf/gf/encoding/gjson"
	"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
}

func NewAccess() (*Access, error) {
	p := NewMQTTProvider()
	return &Access{
		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,
		Cmd:      "report",
	}
	cmdReply := rpcs.ReplySendCommand{}
	return a.SendCommand(cmdArgs, &cmdReply)
}

func (a *Access) SendCommand(args rpcs.ArgsSendCommand, reply *rpcs.ReplySendCommand) error {
	server.Log.Infof("Access Send Command: %v", args)
	cmd := &klink.CloudSend{
		Action:      "cloudSend",
		MsgId:       0,
		DeviceCode:  args.DeviceId,
		SubDeviceId: args.SubDevice,
		Timestamp:   time.Now().Unix(),
		Data: &klink.CloudSendData{
			Cmd:    args.Cmd,
			Params: gjson.New(args.Params),
		},
	}
	msg, err := cmd.Marshal()
	if err != nil {
		return err
	}
	return a.MqttBroker.SendMessageToDevice(args.DeviceId, "c", msg, time.Duration(args.WaitTime)*time.Second)
}