sparrow/services/emqx-agent/agent.go

package main

import (
	"fmt"
	"sparrow/pkg/models"
	"sparrow/pkg/protocol"
	"sparrow/pkg/rpcs"
	"sparrow/pkg/server"
)

type Agent struct {
}

// Message 收到设备上报消息处理
func (a *Agent) Message(topic string, payload []byte) error {
	topicInfo, err := protocol.GetTopicInfo(topic)
	if err != nil {
		return err
	}
	if topicInfo.Direction == protocol.Down {
		return nil
	}
	fmt.Printf("%v", topicInfo)
	return nil
}

// Connected 设备接入时
func (a *Agent) Connected(status *protocol.DevConnectStatus) error {
	// 查询设备信息
	device := &models.Device{}
	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", status.DeviceId, device)
	if err != nil {
		server.Log.Errorf("device not found %s", status.DeviceId)
		return nil
	}
	args := rpcs.ArgsGetOnline{
		Id:                device.RecordId,
		ClientIP:          status.ClientIp,
		AccessRPCHost:     server.GetRPCHost(),
		HeartbeatInterval: 300,
	}
	reply := rpcs.ReplyGetOnline{}
	err = server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetOnlineV2", 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 = device.VendorID
	cArgs.Id = args.Id
	if err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.Online", &cArgs, &cReply); err != nil {
		return err
	}
	return nil
}

// Disconnected 设备断开连接时
func (a *Agent) Disconnected(status *protocol.DevConnectStatus) error {
	// 查询设备信息
	device := &models.Device{}
	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", status.DeviceId, device)
	if err != nil {
		server.Log.Errorf("device not found %s", status.DeviceId)
		return nil
	}
	args := rpcs.ArgsGetOffline{
		Id:       device.RecordId,
		VendorId: device.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", status.DeviceId, err)
	}
	return err
}

func NewAgent() *Agent {
	return &Agent{}
}