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,
	//	SubDevice: 65535,
	//	No:        commandGetCurrentStatus,
	//	Priority:  99,
	//	WaitTime:  0,
	//}
	//cmdReply := rpcs.ReplySendCommand{}
	//err := a.SendCommand(cmdArgs, &cmdReply)
	//if err != nil {
	//	return err
	//}
	//
	//// then wait for status report
	//StatusChan[args.Id] = make(chan *protocol.Data)
	//after := time.After(defaultTimeoutSecond * time.Second)
	//server.Log.Debug("now waiting 5 seconds for status report...")
	//select {
	//case <-after:
	//	// timeout
	//	close(StatusChan[args.Id])
	//	delete(StatusChan, args.Id)
	//	return errors.New("get status timeout.")
	//case data := <-StatusChan[args.Id]:
	//	// go it
	//	close(StatusChan[args.Id])
	//	delete(StatusChan, args.Id)
	//	reply.Status = data.SubData
	//	return nil
	//}
	return nil
}

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