package main import ( "bytes" "context" "encoding/hex" "fmt" "log/slog" "os/signal" "strings" "sync" "syscall" "github.com/AFASystems/presence/internal/pkg/common/appcontext" "github.com/AFASystems/presence/internal/pkg/common/utils" "github.com/AFASystems/presence/internal/pkg/config" "github.com/AFASystems/presence/internal/pkg/kafkaclient" "github.com/AFASystems/presence/internal/pkg/logger" "github.com/AFASystems/presence/internal/pkg/model" "github.com/segmentio/kafka-go" ) var wg sync.WaitGroup func main() { // Load global context to init beacons and latest list appState := appcontext.NewAppState() cfg := config.Load() kafkaManager := kafkaclient.InitKafkaManager() parserRegistry := model.ParserRegistry{ ParserList: make(map[string]model.BeaconParser), } // Set logger -> terminal and log file slog.SetDefault(logger.CreateLogger("decoder.log")) // define context ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) defer stop() readerTopics := []string{"rawbeacons", "parser"} kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "decoder", readerTopics) writerTopics := []string{"alertbeacons"} kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "", writerTopics) slog.Info("Decoder initialized, subscribed to Kafka topics") chRaw := make(chan model.BeaconAdvertisement, 2000) chParser := make(chan model.KafkaParser, 200) wg.Add(3) go kafkaclient.Consume(kafkaManager.GetReader("rawbeacons"), chRaw, ctx, &wg) go kafkaclient.Consume(kafkaManager.GetReader("parser"), chParser, ctx, &wg) eventloop: for { select { case <-ctx.Done(): break eventloop case msg := <-chRaw: processIncoming(msg, appState, kafkaManager.GetWriter("alertbeacons"), &parserRegistry) case msg := <-chParser: switch msg.ID { case "add": config := msg.Config parserRegistry.Register(config.Name, config) case "delete": parserRegistry.Unregister(msg.Name) case "update": config := msg.Config parserRegistry.Register(config.Name, config) } } } slog.Info("broken out of the main event loop") wg.Wait() slog.Info("All go routines have stopped, Beggining to close Kafka connections") kafkaManager.CleanKafkaReaders() kafkaManager.CleanKafkaWriters() } func processIncoming(adv model.BeaconAdvertisement, appState *appcontext.AppState, writer *kafka.Writer, parserRegistry *model.ParserRegistry) { err := decodeBeacon(adv, appState, writer, parserRegistry) if err != nil { eMsg := fmt.Sprintf("Error in decoding: %v", err) fmt.Println(eMsg) return } } func decodeBeacon(adv model.BeaconAdvertisement, appState *appcontext.AppState, writer *kafka.Writer, parserRegistry *model.ParserRegistry) error { beacon := strings.TrimSpace(adv.Data) id := adv.ID if beacon == "" { return nil } b, err := hex.DecodeString(beacon) if err != nil { return err } b = utils.RemoveFlagBytes(b) indeces := utils.ParseADFast(b) event := utils.LoopADStructures(b, indeces, id, parserRegistry) if event.ID == "" { return nil } prevEvent, ok := appState.GetBeaconEvent(id) appState.UpdateBeaconEvent(id, event) if event.Type == "iBeacon" { event.BtnPressed = true } if ok && bytes.Equal(prevEvent.Hash(), event.Hash()) { return nil } eMsg, err := event.ToJSON() if err != nil { return err } if err := writer.WriteMessages(context.Background(), kafka.Message{Value: eMsg}); err != nil { return err } return nil }