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

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
	}

	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.Infof("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 "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
}