sparrow/services/mqttaccess/mqtt_provider.go

package main

import (
	"context"
	"github.com/gogf/gf/encoding/gjson"
	"sparrow/pkg/klink"
	"sparrow/pkg/models"
	"sparrow/pkg/rpcs"
	"sparrow/pkg/server"
	"time"
)

type MQTTProvider struct{}

func NewMQTTProvider() *MQTTProvider {
	return &MQTTProvider{}
}

func (mp *MQTTProvider) ValidateDeviceToken(deviceid string, token []byte) error {
	args := rpcs.ArgsValidateDeviceAccessToken{
		Id:          deviceid,
		AccessToken: token,
	}
	reply := rpcs.ReplyValidateDeviceAccessToken{}
	err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.ValidateDeviceAccessToken", args, &reply)
	if err != nil {
		return err
	}
	return nil
}
func (mp *MQTTProvider) OnDeviceOnline(args rpcs.ArgsGetOnline, VendorId string) error {
	deviceOnlineCount.Inc()
	reply := rpcs.ReplyGetOnline{}
	err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetOnline", args, &reply)
	if err != nil {
		server.Log.Errorf("device online error. args: %v, error: %v", args, err)
	}
	var cReply rpcs.ReplyEmptyResult
	var cArgs rpcs.ArgsGetStatus
	cArgs.VendorId = VendorId
	cArgs.Id = args.Id
	if err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.Online", &cArgs, &cReply); err != nil {
		return err
	}

	// 下发时间同步指令
	var cmdArgs rpcs.ArgsSendCommand
	cmdArgs.Cmd = "timeSync"
	cmdArgs.DeviceId = args.Id
	cmdArgs.SubDevice = "01"
	cmdArgs.Params = map[string]interface{}{
		"value": time.Now().Unix(),
	}
	err = server.RPCCallByName(nil, rpcs.MQTTAccessName, "Access.SendCommand", cmdArgs, &cReply)
	if err != nil {
		server.Log.Errorf("发送时间同步指令：%v", cmdArgs, err)
	}
	return err
}
func (mp *MQTTProvider) OnDeviceOffline(deviceid string, vendorId string) error {
	if deviceid != "" {
		deviceOnlineCount.Dec()
	}
	args := rpcs.ArgsGetOffline{
		Id:       deviceid,
		VendorId: vendorId,
	}
	reply := rpcs.ReplyGetOffline{}
	err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetOffline", args, &reply)
	if err != nil {
		server.Log.Errorf("device offline error. deviceid: %v, error: %v", deviceid, err)
	}
	var cReply rpcs.ReplyEmptyResult
	var cArgs rpcs.ArgsGetStatus
	cArgs.VendorId = vendorId
	cArgs.Id = args.Id
	if err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.Offline", &cArgs, &cReply); err != nil {
		return err
	}

	return err
}
func (mp *MQTTProvider) OnDeviceHeartBeat(deviceid string) error {
	deviceMessageCount.WithLabelValues(deviceid).Inc()
	args := rpcs.ArgsDeviceId{
		Id: deviceid,
	}
	reply := rpcs.ReplyHeartBeat{}
	err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.HeartBeat", args, &reply)
	if err != nil {
		server.Log.Errorf("device heartbeat error. deviceid: %v, error: %v", deviceid, err)
	}
	return err
}
func (mp *MQTTProvider) OnDeviceMessage(deviceid, vendorId string, msgtype string, message *gjson.Json) {
	deviceMessageCount.WithLabelValues(deviceid).Inc()
	server.Log.Debugf("device {%v} message {%v} : %s", deviceid, msgtype, message.MustToJsonString())
	switch msgtype {
	case "s":
		act := klink.PacketAction(message.GetString("action"))
		if act != "" {
			switch act {
			case klink.DevSendAction:
				processReportStatus(deviceid, vendorId, message)
			case klink.DevLoginAction:
				_ = processDevLogin(deviceid, message.GetString("subDeviceId"))
			case klink.DevLogoutAction:
				_ = processDevLogout(deviceid, message.GetString("subDeviceId"))
			case klink.DevNetConfigAction:
				_ = processDevNetConfig(deviceid, message.GetString("md5"))
			case klink.ReportFirmwareAction:
				_ = processDeviceReportUpgrade(deviceid, message.GetString("version"))
			case klink.DevUpgradeAction:
				_ = processDeviceUpgrade(deviceid, message)
			}
		}
	case "e":
		reply := rpcs.ReplyOnEvent{}
		args := rpcs.ArgsOnEvent{
			DeviceId:    deviceid,
			TimeStamp:   message.GetUint64("timestamp"),
			SubDeviceId: message.GetString("subDeviceId"),
			SubData:     message.GetJson("data").MustToJson(),
			VendorId:    vendorId,
		}
		err := server.RPCCallByName(nil, rpcs.ControllerName, "Controller.OnEvent", args, &reply)
		if err != nil {
			server.Log.Errorf("device on event error. args: %v, error: %v", args, err)
			return
		}
	default:
		server.Log.Infof("unknown message type: %v", msgtype)
	}
}

func processReportStatus(deviceid, vendorId string, message *gjson.Json) {
	reply := rpcs.ReplyOnStatus{}
	args := rpcs.ArgsOnStatus{
		DeviceId:    deviceid,
		Timestamp:   message.GetUint64("timestamp"),
		SubData:     message.GetJson("data").MustToJson(),
		VendorId:    vendorId,
		SubDeviceId: message.GetString("subDeviceId"),
		Action:      klink.PacketAction(message.GetString("action")),
	}
	err := server.RPCCallByName(nil, rpcs.ControllerName, "Controller.OnStatus", args, &reply)
	if err != nil {
		server.Log.Errorf("device report status error. args: %v, error: %v", args, err)
		return
	}
}

func processDevLogin(deviceCode, subDeviceId string) error {
	var args rpcs.SubDeviceArgs
	args.DeviceCode = deviceCode
	args.Status = 1
	args.SubDeviceId = subDeviceId
	var reply *models.SubDevice
	err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.UpdateSubDevice", &args, &reply)
	if err != nil {
		server.Log.Errorf("子设备上线出错：%s", err.Error())
	}
	return nil
}

func processDevLogout(deviceCode, subDeviceId string) error {
	var args rpcs.SubDeviceArgs
	args.DeviceCode = deviceCode
	args.Status = 0
	args.SubDeviceId = subDeviceId
	var reply *models.SubDevice
	err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.UpdateSubDevice", &args, &reply)
	if err != nil {
		server.Log.Errorf("子设备下线出错：%s", err.Error())
	}
	return nil
}

// 处理设备配网信息
func processDevNetConfig(deviceCode, md5 string) error {
	args := &models.DeviceNetConfig{
		DeviceIdentifier: deviceCode,
		MD5:              md5,
		Status:           1,
	}
	reply := rpcs.ReplyCheckDeviceNetConfig{}
	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.CreateDeviceNetConfig", args, &reply)
	if err != nil {
		server.Log.Errorf("set device:%s net config info error:%v", deviceCode, err)
	}
	return nil
}

// 设备上报固件信息处理
func processDeviceReportUpgrade(deviceId, version string) error {
	args := &rpcs.ArgsUpdateDeviceVersion{
		DeviceId: deviceId,
		Version:  version,
	}
	var reply rpcs.ReplyEmptyResult
	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.UpdateDeviceVersion", args, &reply)
	if err != nil {
		server.Log.Errorf("更新设备版本号失败：%v", err)
	}
	return nil
}

func processDeviceUpgrade(deviceId string, message *gjson.Json) error {
	var reply rpcs.ReplyEmptyResult
	data := gjson.New(message.GetJson("data").MustToJson())
	switch data.GetString("cmd") {
	case "download":
		params := gjson.New(data.GetJson("params").MustToJson())

		args := &rpcs.ChunkUpgrade{
			DeviceId: deviceId,
			FileId:   params.GetString("fileId"),
			FileSize: params.GetInt64("fileSize"),
			Size:     params.GetInt64("size"),
			Offset:   params.GetInt("offset"),
		}

		err := server.RPCCallByName(nil, rpcs.MQTTAccessName, "Access.ChunkUpgrade", args, &reply)
		if err != nil {
			server.Log.Errorf("分片下载发送失败：%v", err)
		}

	case "downProgress":
		params := gjson.New(data.GetJson("params").MustToJson())
		var args rpcs.ArgsOtaProgress
		args.DeviceId = deviceId
		args.Progress = params.GetInt("progress")

		err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetProgress", args, &reply)
		if err != nil {
			server.Log.Errorf("OTA升级进度获取失败：%v", err)
			return err
		}
	}

	return nil
}