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(), 60*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 }