evioEvents.go 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171
  1. package networkEvents
  2. import (
  3. "context"
  4. "fmt"
  5. "fwh-toolbox-server/logger"
  6. "fwh-toolbox-server/utils"
  7. "sync"
  8. "time"
  9. "github.com/tidwall/evio"
  10. "go.uber.org/zap"
  11. )
  12. type SiteInfo struct {
  13. Sn string `json:"sn"`
  14. Uuid string `json:"uuid"`
  15. }
  16. var connMap sync.Map // key - string UUID, value - evio.Conn
  17. type ConnectContext = utils.ConnectContext
  18. func InitEvio() error {
  19. var events evio.Events
  20. events.Opened = EventsOpened
  21. events.Closed = EventsClosed
  22. events.Data = EventsData
  23. if err := evio.Serve(events, "tcp://0.0.0.0:9099"); err != nil {
  24. return err
  25. }
  26. return nil
  27. }
  28. func EventsOpened(c evio.Conn) (out []byte, opts evio.Options, action evio.Action) {
  29. connectContext := &ConnectContext{
  30. UUID: "",
  31. SN: "",
  32. XlogPackageParser: nil,
  33. DoneChan: make(chan string, 1), // 缓冲避免阻塞
  34. }
  35. connectContext.XlogPackageParser = utils.NewXlogPackageParser(connectContext)
  36. connectContext.SendBuffer.Store([]byte(nil))
  37. c.SetContext(connectContext)
  38. // connMap.Store(c.RemoteAddr().String(), c) // 保存连接
  39. // xlogPackageParserMap.Store(c.RemoteAddr().String(), utils.NewXlogPackageParser())
  40. opts = evio.Options{}
  41. action = evio.None
  42. return
  43. }
  44. func EventsClosed(c evio.Conn, err error) (action evio.Action) {
  45. if connectContext := c.Context().(*ConnectContext); connectContext != nil {
  46. uuid := connectContext.UUID
  47. connMap.Delete(uuid)
  48. logger.Std.Println("断开连接端口 UUID: ", uuid)
  49. }
  50. // xlogPackageParserMap.Delete(c.RemoteAddr().String())
  51. action = evio.None
  52. return
  53. }
  54. func EventsData(c evio.Conn, in []byte) (out []byte, action evio.Action) {
  55. connectContext, ok := c.Context().(*ConnectContext)
  56. if !ok || connectContext == nil {
  57. logger.Std.Println("EventsData Context invalid.")
  58. logger.Std.Println(connectContext)
  59. action = evio.None
  60. return
  61. }
  62. uuid := connectContext.UUID
  63. logger.Std.Println("接收数据 UUID: ", uuid)
  64. out = nil
  65. if in == nil {
  66. // 处理数据发送事件
  67. buffer := connectContext.SendBuffer.Swap([]byte(nil)).([]byte)
  68. if len(buffer) > 0 {
  69. out = buffer
  70. }
  71. } else {
  72. // 处理数据接收事件
  73. parser := connectContext.XlogPackageParser
  74. if !parser.Handshake {
  75. data, err := parser.ParseXlogHandshakePackage(in)
  76. if err != nil {
  77. zap.L().Error("握手失败", zap.String("err", err.Error()))
  78. } else {
  79. out, err = parser.CreateXlogHandshakePackage()
  80. if err != nil {
  81. zap.L().Error("握手失败", zap.String("err", err.Error()))
  82. } else {
  83. parser.Handshake = true
  84. connectContext.UUID = string(data.DevUuid[:])
  85. connMap.Store(string(data.DevUuid[:]), c) // 保存连接
  86. zap.L().Info("握手校验", zap.Binary("newIv", data.NewIv))
  87. }
  88. }
  89. } else {
  90. parser.PreparserXlogPackage(in)
  91. }
  92. }
  93. action = evio.None
  94. return
  95. }
  96. func GetConnectList() []SiteInfo {
  97. infoList := make([]SiteInfo, 0)
  98. connMap.Range(func(key, value any) bool {
  99. conn := value.(evio.Conn)
  100. ctx := conn.Context().(*ConnectContext)
  101. infoList = append(infoList, SiteInfo{
  102. Sn: ctx.SN,
  103. Uuid: key.(string),
  104. })
  105. return true
  106. })
  107. return infoList
  108. }
  109. func SendXlogCmdDate(uuid string, data []byte) (string, error) {
  110. var err error
  111. if val, ok := connMap.Load(uuid); ok {
  112. c := val.(evio.Conn)
  113. connectContext, ok := c.Context().(*ConnectContext)
  114. if !ok || connectContext == nil {
  115. err = fmt.Errorf("send xlog cmd filed : Context missing, UUID: %s", uuid)
  116. return "", err
  117. }
  118. buffer, _ := connectContext.SendBuffer.Load().([]byte)
  119. if buffer != nil {
  120. err = fmt.Errorf("send xlog cmd filed : Data sending, UUID: %s", uuid)
  121. return "", err
  122. }
  123. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  124. defer cancel()
  125. // 创建一个带缓冲的 done channel
  126. done := make(chan string, 1)
  127. connectContext.DoneChan = done
  128. // 存储要发送的数据
  129. cmdPackage, err := connectContext.XlogPackageParser.CreateXlogCmdPackage(append([]byte(connectContext.UUID)[len(uuid)-8:], data...))
  130. if err != nil {
  131. return "", err
  132. }
  133. connectContext.SendBuffer.Store(cmdPackage)
  134. // 唤醒 evio,让它触发 Data 事件去真正写数据
  135. c.Wake()
  136. // 等待结果或超时
  137. select {
  138. case result := <-done:
  139. return result, nil
  140. case <-ctx.Done():
  141. return "", fmt.Errorf("send xlog cmd timeout, UUID: %s, err: %v", uuid, ctx.Err())
  142. }
  143. } else {
  144. err = fmt.Errorf("send xlog cmd filed : Connect missing, UUID: %s", uuid)
  145. }
  146. return "", err
  147. }