evioEvents.go 4.4 KB

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