| @@ -3,6 +3,7 @@ package bridge | |||||
| import ( | import ( | ||||
| "context" | "context" | ||||
| "encoding/json" | "encoding/json" | ||||
| "fmt" | |||||
| "log/slog" | "log/slog" | ||||
| "sync" | "sync" | ||||
| "time" | "time" | ||||
| @@ -96,6 +97,7 @@ func (a *BridgeApp) Run(ctx context.Context) { | |||||
| case msg := <-a.ChApi: | case msg := <-a.ChApi: | ||||
| switch msg.Method { | switch msg.Method { | ||||
| case "POST": | case "POST": | ||||
| fmt.Println("adding beacon to lookup", "mac", msg.MAC, "id", msg.ID) | |||||
| a.AppState.AddBeaconToLookup(msg.MAC, msg.ID) | a.AppState.AddBeaconToLookup(msg.MAC, msg.ID) | ||||
| slog.Info("beacon added to lookup", "id", msg.ID) | slog.Info("beacon added to lookup", "id", msg.ID) | ||||
| case "DELETE": | case "DELETE": | ||||
| @@ -2,6 +2,7 @@ package decoder | |||||
| import ( | import ( | ||||
| "context" | "context" | ||||
| "fmt" | |||||
| "log/slog" | "log/slog" | ||||
| "sync" | "sync" | ||||
| "time" | "time" | ||||
| @@ -71,7 +72,6 @@ func (a *DecoderApp) Run(ctx context.Context) { | |||||
| return | return | ||||
| case <-healthTicker.C: | case <-healthTicker.C: | ||||
| health, err := a.AppState.GetDecoderHealth(a.KafkaManager) | health, err := a.AppState.GetDecoderHealth(a.KafkaManager) | ||||
| slog.Info("decoder health", "health", string(health)) | |||||
| if err != nil { | if err != nil { | ||||
| slog.Error("getting decoder health", "err", err) | slog.Error("getting decoder health", "err", err) | ||||
| continue | continue | ||||
| @@ -84,6 +84,7 @@ func (a *DecoderApp) Run(ctx context.Context) { | |||||
| continue | continue | ||||
| } | } | ||||
| case msg := <-a.ChRaw: | case msg := <-a.ChRaw: | ||||
| fmt.Println("msg: ", msg) | |||||
| decoder.ProcessIncoming(msg, a.AppState, a.KafkaManager.GetWriter("alertbeacons"), a.ParserRegistry) | decoder.ProcessIncoming(msg, a.AppState, a.KafkaManager.GetWriter("alertbeacons"), a.ParserRegistry) | ||||
| case msg := <-a.ChParser: | case msg := <-a.ChParser: | ||||
| switch msg.ID { | switch msg.ID { | ||||
| @@ -3,6 +3,7 @@ package bridge | |||||
| import ( | import ( | ||||
| "context" | "context" | ||||
| "encoding/json" | "encoding/json" | ||||
| "fmt" | |||||
| "log/slog" | "log/slog" | ||||
| "strings" | "strings" | ||||
| "time" | "time" | ||||
| @@ -13,15 +14,10 @@ import ( | |||||
| "github.com/segmentio/kafka-go" | "github.com/segmentio/kafka-go" | ||||
| ) | ) | ||||
| // BeaconLookup provides MAC->ID lookup (e.g. AppState). | |||||
| type BeaconLookup interface { | |||||
| BeaconExists(mac string) (id string, ok bool) | |||||
| } | |||||
| // HandleMQTTMessage processes an MQTT message: parses JSON array of RawReading or CSV. | // HandleMQTTMessage processes an MQTT message: parses JSON array of RawReading or CSV. | ||||
| // For JSON, converts each reading to BeaconAdvertisement and writes to the writer if MAC is in lookup. | // For JSON, converts each reading to BeaconAdvertisement and writes to the writer if MAC is in lookup. | ||||
| // Hostname is derived from topic (e.g. "publish_out/gateway1" -> "gateway1"). Safe if topic has no "/". | // Hostname is derived from topic (e.g. "publish_out/gateway1" -> "gateway1"). Safe if topic has no "/". | ||||
| func HandleMQTTMessage(topic string, payload []byte, lookup BeaconLookup, writer *kafka.Writer) { | |||||
| func HandleMQTTMessage(topic string, payload []byte, appState *appcontext.AppState, writer *kafka.Writer) { | |||||
| parts := strings.SplitN(topic, "/", 2) | parts := strings.SplitN(topic, "/", 2) | ||||
| hostname := "" | hostname := "" | ||||
| if len(parts) >= 2 { | if len(parts) >= 2 { | ||||
| @@ -39,10 +35,11 @@ func HandleMQTTMessage(topic string, payload []byte, lookup BeaconLookup, writer | |||||
| if reading.Type == "Gateway" { | if reading.Type == "Gateway" { | ||||
| continue | continue | ||||
| } | } | ||||
| id, ok := lookup.BeaconExists(reading.MAC) | |||||
| id, ok := appState.BeaconExists(reading.MAC) | |||||
| if !ok { | if !ok { | ||||
| continue | continue | ||||
| } | } | ||||
| fmt.Println("beacon found: ", id) | |||||
| adv := appcontext.BeaconAdvertisement{ | adv := appcontext.BeaconAdvertisement{ | ||||
| ID: id, | ID: id, | ||||
| Hostname: hostname, | Hostname: hostname, | ||||
| @@ -30,6 +30,8 @@ func DecodeBeacon(adv appcontext.BeaconAdvertisement, appState *appcontext.AppSt | |||||
| return nil | return nil | ||||
| } | } | ||||
| fmt.Println("beacon: ", beacon) | |||||
| b, err := hex.DecodeString(beacon) | b, err := hex.DecodeString(beacon) | ||||
| if err != nil { | if err != nil { | ||||
| return err | return err | ||||
| @@ -18,7 +18,6 @@ import ( | |||||
| ) | ) | ||||
| func findTracker(msg model.HTTPLocation, db *gorm.DB) (model.Tracker, error) { | func findTracker(msg model.HTTPLocation, db *gorm.DB) (model.Tracker, error) { | ||||
| fmt.Printf("Finding tracker for MAC: %s, ID: %s\n", msg.MAC, msg.ID) | |||||
| var tracker model.Tracker | var tracker model.Tracker | ||||
| if msg.MAC != "" { | if msg.MAC != "" { | ||||
| if err := db.Where("mac = ?", strings.ToUpper(strings.ReplaceAll(msg.MAC, ":", ""))).Find(&tracker).Error; err != nil { | if err := db.Where("mac = ?", strings.ToUpper(strings.ReplaceAll(msg.MAC, ":", ""))).Find(&tracker).Error; err != nil { | ||||