sub_dev.go 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211
  1. package main
  2. import (
  3. "context"
  4. "encoding/json"
  5. mqtt "github.com/eclipse/paho.mqtt.golang"
  6. "github.com/gogf/gf/os/grpool"
  7. "runtime"
  8. "runtime/debug"
  9. "sparrow/pkg/protocol"
  10. "sparrow/pkg/server"
  11. "sparrow/services/emqx-agent/client"
  12. "strings"
  13. "time"
  14. )
  15. type SubDev interface {
  16. SubDevMsg(handle Handle) error
  17. PublishToMsgToDev(topic string, payload []byte) error
  18. }
  19. type ConnectMsg struct {
  20. Username string `json:"username"`
  21. Ts int64 `json:"ts"`
  22. Sockport int `json:"sockport"`
  23. ProtoVer int `json:"proto_ver"`
  24. ProtoName string `json:"proto_name"`
  25. Keepalive int `json:"keepalive"`
  26. Ipaddress string `json:"ipaddress"`
  27. ExpiryInterval int `json:"expiry_interval"`
  28. ConnectedAt int64 `json:"connected_at"`
  29. Connack int `json:"connack"`
  30. Clientid string `json:"clientid"`
  31. Reason string `json:"reason"`
  32. CleanStart bool `json:"clean_start"`
  33. }
  34. type Handle func(ctx context.Context) DevSubHandle
  35. type DevSubHandle interface {
  36. Message(topic string, payload []byte) error
  37. Connected(status *protocol.DevConnectStatus) error
  38. Disconnected(status *protocol.DevConnectStatus) error
  39. }
  40. type MqttClient struct {
  41. client *client.MqttClient
  42. handlePool *grpool.Pool
  43. }
  44. func (d *MqttClient) PublishToMsgToDev(topic string, payload []byte) error {
  45. return d.client.Publish(topic, 1, false, payload)
  46. }
  47. const (
  48. ShareSubTopicPrefix = "$share/sparrow.agent/"
  49. TopicConnectStatus = ShareSubTopicPrefix + "$SYS/brokers/+/clients/#"
  50. TopicThing = ShareSubTopicPrefix + protocol.TopicHeadThing + "/up/#"
  51. TopicOta = ShareSubTopicPrefix + protocol.TopicHeadOta + "/up/#"
  52. TopicConfig = ShareSubTopicPrefix + protocol.TopicHeadConfig + "/up/#"
  53. TopicSDKLog = ShareSubTopicPrefix + protocol.TopicHeadLog + "/up/#"
  54. TopicShadow = ShareSubTopicPrefix + protocol.TopicHeadShadow + "/up/#"
  55. TopicGateway = ShareSubTopicPrefix + protocol.TopicHeadGateway + "/up/#"
  56. TopicExt = ShareSubTopicPrefix + protocol.TopicHeadExt + "/up/#"
  57. TopicEvent = ShareSubTopicPrefix + protocol.TopicHeadEvent + "/up/#"
  58. )
  59. func NewSubDev(conf *client.MqttConfig) (SubDev, error) {
  60. return newEmqClient(conf)
  61. }
  62. func newEmqClient(conf *client.MqttConfig) (SubDev, error) {
  63. mc, err := client.NewMqttClient(conf)
  64. if err != nil {
  65. return nil, err
  66. }
  67. return &MqttClient{
  68. client: mc,
  69. handlePool: grpool.New(1000),
  70. }, nil
  71. }
  72. func (d *MqttClient) SubDevMsg(handle Handle) error {
  73. err := d.subDevMsg(nil, handle)
  74. if err != nil {
  75. return err
  76. }
  77. client.SetMqttSetOnConnectHandler(func(cli mqtt.Client) {
  78. err := d.subDevMsg(cli, handle)
  79. if err != nil {
  80. server.Log.Errorf("mqttSetOnConnectHandler.subDevMsg err:%v", err)
  81. }
  82. })
  83. return nil
  84. }
  85. func (d *MqttClient) subDevMsg(cli mqtt.Client, handle Handle) error {
  86. err := d.subscribeWithFunc(cli, TopicConnectStatus, d.subConnectStatus(handle))
  87. if err != nil {
  88. return err
  89. }
  90. err = d.subscribeWithFunc(cli, TopicThing, func(ctx context.Context, topic string, payload []byte) error {
  91. return handle(ctx).Message(topic, payload)
  92. })
  93. if err != nil {
  94. return err
  95. }
  96. err = d.subscribeWithFunc(cli, TopicConfig, func(ctx context.Context, topic string, payload []byte) error {
  97. return handle(ctx).Message(topic, payload)
  98. })
  99. if err != nil {
  100. return err
  101. }
  102. err = d.subscribeWithFunc(cli, TopicOta, func(ctx context.Context, topic string, payload []byte) error {
  103. return handle(ctx).Message(topic, payload)
  104. })
  105. if err != nil {
  106. return err
  107. }
  108. err = d.subscribeWithFunc(cli, TopicExt, func(ctx context.Context, topic string, payload []byte) error {
  109. return handle(ctx).Message(topic, payload)
  110. })
  111. if err != nil {
  112. return err
  113. }
  114. err = d.subscribeWithFunc(cli, TopicShadow, func(ctx context.Context, topic string, payload []byte) error {
  115. return handle(ctx).Message(topic, payload)
  116. })
  117. if err != nil {
  118. return err
  119. }
  120. err = d.subscribeWithFunc(cli, TopicGateway, func(ctx context.Context, topic string, payload []byte) error {
  121. return handle(ctx).Message(topic, payload)
  122. })
  123. if err != nil {
  124. return err
  125. }
  126. err = d.subscribeWithFunc(cli, TopicSDKLog, func(ctx context.Context, topic string, payload []byte) error {
  127. return handle(ctx).Message(topic, payload)
  128. })
  129. if err != nil {
  130. return err
  131. }
  132. err = d.subscribeWithFunc(cli, TopicEvent, func(ctx context.Context, topic string, payload []byte) error {
  133. return handle(ctx).Message(topic, payload)
  134. })
  135. if err != nil {
  136. return err
  137. }
  138. return nil
  139. }
  140. func (d *MqttClient) subConnectStatus(handle Handle) func(ctx context.Context, topic string, payload []byte) error {
  141. return func(ctx context.Context, topic string, payload []byte) error {
  142. var (
  143. msg ConnectMsg
  144. err error
  145. )
  146. err = json.Unmarshal(payload, &msg)
  147. if err != nil {
  148. server.Log.Errorf("json.Unmarshal err :%s, topic :%v", err, topic)
  149. return err
  150. }
  151. status := protocol.DevConnectStatus{
  152. DeviceCode: msg.Username,
  153. DeviceId: msg.Clientid,
  154. ClientIp: msg.Ipaddress,
  155. }
  156. if strings.HasSuffix(topic, "/connected") || strings.HasSuffix(topic, "/subscribed") {
  157. status.Action = "LOGIN"
  158. return handle(ctx).Connected(&status)
  159. } else {
  160. status.Action = "LOGOUT"
  161. status.Reason = msg.Reason
  162. return handle(ctx).Disconnected(&status)
  163. }
  164. }
  165. }
  166. func (d *MqttClient) subscribeWithFunc(cli mqtt.Client, topic string,
  167. handle func(ctx context.Context, topic string, payload []byte) error) error {
  168. return d.client.Subscribe(cli, topic, 1, func(c mqtt.Client, message mqtt.Message) {
  169. _ = d.handlePool.Add(func() {
  170. go func() {
  171. ctx, cancel := context.WithTimeout(context.Background(), 50*time.Second)
  172. defer cancel()
  173. Recover(ctx)
  174. err := handle(ctx, message.Topic(), message.Payload())
  175. if err != nil {
  176. server.Log.Errorf("handle failure err :%s, topic :%v", err, topic)
  177. }
  178. }()
  179. })
  180. })
  181. }
  182. func Recover(ctx context.Context) {
  183. if p := recover(); p != nil {
  184. HandleThrow(ctx, p)
  185. }
  186. }
  187. func HandleThrow(ctx context.Context, p any) {
  188. pc := make([]uintptr, 1)
  189. runtime.Callers(3, pc)
  190. f := runtime.FuncForPC(pc[0])
  191. server.Log.Errorf("HandleThrow|func=%s|error=%#v|stack=%s\n", f, p, string(debug.Stack()))
  192. //os.Exit(-1)
  193. }