You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

109 lines
2.4 KiB

  1. package mqtthandler
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "log"
  7. "os"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/AFASystems/presence/internal/pkg/model"
  12. "github.com/segmentio/kafka-go"
  13. )
  14. func MqttHandler(writer *kafka.Writer, topicName []byte, message []byte) {
  15. hostname := strings.Split(string(topicName), "/")[1]
  16. msgStr := string(message)
  17. if strings.HasPrefix(msgStr, "[") {
  18. var readings []model.RawReading
  19. err := json.Unmarshal(message, &readings)
  20. if err != nil {
  21. log.Printf("Error parsing JSON: %v", err)
  22. return
  23. }
  24. for _, reading := range readings {
  25. if reading.Type == "Gateway" {
  26. continue
  27. }
  28. adv := model.BeaconAdvertisement{
  29. Hostname: hostname,
  30. MAC: reading.MAC,
  31. RSSI: int64(reading.RSSI),
  32. Data: reading.RawData,
  33. HSButtonCounter: parseButtonState(reading.RawData),
  34. }
  35. encodedMsg, err := json.Marshal(adv)
  36. if err != nil {
  37. fmt.Println("Error in marshaling: ", err)
  38. break
  39. }
  40. msg := kafka.Message{
  41. Value: encodedMsg,
  42. }
  43. err = writer.WriteMessages(context.Background(), msg)
  44. if err != nil {
  45. fmt.Println("Error in writing to Kafka: ", err)
  46. time.Sleep(1 * time.Second)
  47. break
  48. }
  49. fmt.Println("message sent: ", time.Now())
  50. }
  51. } else {
  52. s := strings.Split(string(message), ",")
  53. if len(s) < 6 {
  54. log.Printf("Messaggio CSV non valido: %s", msgStr)
  55. return
  56. }
  57. rawdata := s[4]
  58. buttonCounter := parseButtonState(rawdata)
  59. if buttonCounter > 0 {
  60. adv := model.BeaconAdvertisement{}
  61. i, _ := strconv.ParseInt(s[3], 10, 64)
  62. adv.Hostname = hostname
  63. adv.BeaconType = "hb_button"
  64. adv.MAC = s[1]
  65. adv.RSSI = i
  66. adv.Data = rawdata
  67. adv.HSButtonCounter = buttonCounter
  68. read_line := strings.TrimRight(string(s[5]), "\r\n")
  69. it, err33 := strconv.Atoi(read_line)
  70. if err33 != nil {
  71. fmt.Println(it)
  72. fmt.Println(err33)
  73. os.Exit(2)
  74. }
  75. }
  76. }
  77. }
  78. func parseButtonState(raw string) int64 {
  79. raw = strings.ToUpper(raw)
  80. if strings.HasPrefix(raw, "0201060303E1FF12") && len(raw) >= 38 {
  81. buttonField := raw[34:38]
  82. if buttonValue, err := strconv.ParseInt(buttonField, 16, 64); err == nil {
  83. return buttonValue
  84. }
  85. }
  86. if strings.HasPrefix(raw, "02010612FF590") && len(raw) >= 24 {
  87. counterField := raw[22:24]
  88. buttonState, err := strconv.ParseInt(counterField, 16, 64)
  89. if err == nil {
  90. return buttonState
  91. }
  92. }
  93. return 0
  94. }