package networkEvents import ( "context" "fmt" "fwh-toolbox-server/logger" "fwh-toolbox-server/utils" "sync" "time" "github.com/tidwall/evio" "go.uber.org/zap" ) type SiteInfo struct { Sn string `json:"sn"` Uuid string `json:"uuid"` } var connMap sync.Map // key - string UUID, value - evio.Conn type ConnectContext = utils.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) { connectContext := &ConnectContext{ UUID: "", SN: "", XlogPackageParser: nil, DoneChan: make(chan string, 1), // 缓冲避免阻塞 } connectContext.XlogPackageParser = utils.NewXlogPackageParser(connectContext) connectContext.SendBuffer.Store([]byte(nil)) c.SetContext(connectContext) // connMap.Store(c.RemoteAddr().String(), c) // 保存连接 // xlogPackageParserMap.Store(c.RemoteAddr().String(), utils.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 } uuid := connectContext.UUID logger.Std.Println("接收数据 UUID: ", uuid) 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 { 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 connectContext.UUID = string(data.DevUuid[:]) connMap.Store(string(data.DevUuid[:]), c) // 保存连接 zap.L().Info("握手校验", zap.Binary("newIv", data.NewIv)) } } } else { parser.PreparserXlogPackage(in) } } action = evio.None return } func GetConnectList() []SiteInfo { infoList := make([]SiteInfo, 0) connMap.Range(func(key, value any) bool { conn := value.(evio.Conn) ctx := conn.Context().(*ConnectContext) infoList = append(infoList, SiteInfo{ Sn: ctx.SN, Uuid: key.(string), }) 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 } 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 // 存储要发送的数据 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: 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 }