FranklinWH-Toolbox/service/networkEvents/evioEvents.go

package networkEvents

import (
	"context"
	"fmt"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/logger"
	"fwh-toolbox-server/utils/xlog"
	"regexp"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/tidwall/evio"
	"go.uber.org/zap"
)

var connMap sync.Map // key - string UUID, value - evio.Conn

type ConnectContext = xlog.ConnectContext

func InitEvio() error {
	var events evio.Events
	events.Opened = EventsOpened
	events.Closed = EventsClosed
	events.Data = EventsData
	if err := evio.Serve(events, "tcp://0.0.0.0:9099"); err != nil {
		return err
	}
	return nil
}

func EventsOpened(c evio.Conn) (out []byte, opts evio.Options, action evio.Action) {
	logger.Std.Println("EventsOpened ", c.RemoteAddr().String())
	newUUID := uuid.New().String()
	connectContext := &ConnectContext{
		UUID:              newUUID,
		SN:                "",
		Version:           "xlog v1.0",
		XlogPackageParser: nil,
		DoneChan:          make(chan string, 1), // 缓冲避免阻塞
		ChanIsRunning:     false,
	}
	connectContext.XlogPackageParser = xlog.NewXlogPackageParser(connectContext)
	connectContext.SendBuffer.Store([]byte(nil))
	c.SetContext(connectContext)

	connMap.Store(newUUID, c) // 预保存连接（兼容xlog 1.0）

	// connMap.Store(c.RemoteAddr().String(), c) // 保存连接
	// xlogPackageParserMap.Store(c.RemoteAddr().String(), xlog.NewXlogPackageParser())
	opts = evio.Options{}
	action = evio.None
	return
}

func EventsClosed(c evio.Conn, err error) (action evio.Action) {
	if connectContext := c.Context().(*ConnectContext); connectContext != nil {
		uuid := connectContext.UUID
		connMap.Delete(uuid)
		logger.Std.Println("断开连接端口 UUID: ", uuid)
	}

	// xlogPackageParserMap.Delete(c.RemoteAddr().String())
	action = evio.None
	return
}

func EventsData(c evio.Conn, in []byte) (out []byte, action evio.Action) {
	connectContext, ok := c.Context().(*ConnectContext)
	if !ok || connectContext == nil {
		logger.Std.Println("EventsData Context invalid.")
		logger.Std.Println(connectContext)
		action = evio.None
		return
	}

	// logger.Std.Println("接收数据 UUID: ", connectContext.UUID)
	// logger.Std.Println("接收数据: ", in)

	out = nil

	if in == nil {
		// 处理数据发送事件
		buffer := connectContext.SendBuffer.Swap([]byte(nil)).([]byte)
		if len(buffer) > 0 {
			out = buffer
		}
	} else {
		// 处理数据接收事件
		parser := connectContext.XlogPackageParser
		if !parser.Handshake {
			// 先判断协议版本，是否是xlog1.0
			ok, err := parser.ParseXlogCheckVersionIsOld(in)
			if err != nil {
				zap.L().Error("版本判断失败", zap.String("err", err.Error()))
			} else {
				if ok {
					parser.Handshake = true
					parser.PreparserXlog10Package(in)
				} else {
					data, err := parser.ParseXlogHandshakePackage(in)
					if err != nil {
						zap.L().Error("握手失败", zap.String("err", err.Error()))
					} else {
						out, err = parser.CreateXlogHandshakePackage()
						if err != nil {
							zap.L().Error("握手失败", zap.String("err", err.Error()))
						} else {
							parser.Handshake = true
							parser.IsOldProtocol = false

							connMap.Delete(connectContext.UUID)           // 删除预保存的连接，使用xlog v2.0的UUID
							connectContext.UUID = string(data.DevUuid[:]) // 覆盖UUID
							connectContext.Version = "xlog v2.0"
							connMap.Store(connectContext.UUID, c) // 保存xlog v2.0的连接

							zap.L().Info("握手校验", zap.Binary("newIv", data.NewIv[:]))
						}
					}
				}
			}
		} else if parser.IsOldProtocol {
			parser.PreparserXlog10Package(in)
		} else {
			parser.PreparserXlogPackage(in)
		}
	}

	action = evio.None
	return
}

func GetConnectList() []domain.XlogSiteInfo {
	infoList := make([]domain.XlogSiteInfo, 0)
	connMap.Range(func(key, value any) bool {
		conn := value.(evio.Conn)
		ctx := conn.Context().(*ConnectContext)
		infoList = append(infoList, domain.XlogSiteInfo{
			Sn:      ctx.SN,
			Uuid:    key.(string),
			Version: ctx.Version,
		})
		return true
	})
	return infoList
}

func SendXlogCmdDate(uuid string, data []byte) (string, error) {
	var err error
	if val, ok := connMap.Load(uuid); ok {
		c := val.(evio.Conn)
		connectContext, ok := c.Context().(*ConnectContext)
		if !ok || connectContext == nil {
			err = fmt.Errorf("send xlog cmd filed : Context missing, UUID: %s", uuid)
			return "", err
		}

		// 判断连接是否在传输结果
		if connectContext.ChanIsRunning {
			return "命令正在执行中，请稍后...", nil
		}
		connectContext.ChanIsRunning = true
		defer func() {
			connectContext.ChanIsRunning = false
		}()

		buffer, _ := connectContext.SendBuffer.Load().([]byte)
		if buffer != nil {
			err = fmt.Errorf("send xlog cmd filed : Data sending, UUID: %s", uuid)
			return "", err
		}

		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()

		// 创建一个带缓冲的 done channel
		done := make(chan string, 1)
		connectContext.DoneChan = done

		// 存储要发送的数据
		var cmdPackage []byte
		if connectContext.XlogPackageParser.IsOldProtocol {
			cmdPackage, err = connectContext.XlogPackageParser.CreateXlog10CmdPackage(data)
		} else {
			cmdPackage, err = connectContext.XlogPackageParser.CreateXlogCmdPackage(append([]byte(connectContext.UUID)[len(uuid)-8:], data...))
		}
		if err != nil {
			return "", err
		}
		connectContext.SendBuffer.Store(cmdPackage)
		// 唤醒 evio，让它触发 Data 事件去真正写数据
		c.Wake()

		// 等待结果或超时
		select {
		case result := <-done:
			if connectContext.XlogPackageParser.IsOldProtocol && len(connectContext.SN) == 0 {
				re := regexp.MustCompile(`"AGATE_SN"\s*:\s*"([^"]*)"`)
				match := re.FindStringSubmatch(result)
				if len(match) > 1 {
					fmt.Println("AGATE_SN:", match[1])
					connectContext.SN = match[1]
				}
			}
			return result, nil
		case <-ctx.Done():
			return "", fmt.Errorf("send xlog cmd timeout, UUID: %s, err: %v", uuid, ctx.Err())
		}
	} else {
		err = fmt.Errorf("send xlog cmd filed : Connect missing, UUID: %s", uuid)
	}
	return "", err
}