diff --git a/internal/nats/sub.go b/internal/nats/sub.go index 5e07d23..f79e49d 100644 --- a/internal/nats/sub.go +++ b/internal/nats/sub.go @@ -9,6 +9,7 @@ import ( "context" "errors" "fmt" + "sync" "github.com/ThreeDotsLabs/watermill" "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats" @@ -17,6 +18,7 @@ import ( ) type NatsHandler struct { + mu sync.Mutex config *config.Config hasuraRepo *repository.HasuraRepository } @@ -63,6 +65,8 @@ func (n *NatsHandler) Subscribe() { } } func (n *NatsHandler) handleMessage(msg *message.Message) error { + n.mu.Lock() + defer n.mu.Unlock() log := utils.GetSugaredLogger() if msg.Payload == nil { // fmt.Println("empty message") @@ -83,9 +87,11 @@ func (n *NatsHandler) handleMessage(msg *message.Message) error { log.Infof("parsed [%s]: %v\n", msg.UUID, parsed) } - parsed.Uuid = msg.UUID - if err = n.hasuraRepo.CreateNew(parsed); err != nil { - fmt.Print("error inserting message", err) + if parsed != nil { + parsed.Uuid = msg.UUID + if err = n.hasuraRepo.CreateNew(parsed); err != nil { + fmt.Print("error inserting message", err) + } } return err } diff --git a/internal/parsers/aviation_parser.go b/internal/parsers/aviation_parser.go index eb99f8d..563960d 100644 --- a/internal/parsers/aviation_parser.go +++ b/internal/parsers/aviation_parser.go @@ -3,9 +3,11 @@ package parsers import ( "caatsm/internal/config" "caatsm/internal/domain" + "caatsm/pkg/utils" "fmt" "regexp" "strings" + "sync" "time" ) @@ -30,27 +32,28 @@ var ( otherPatterns = []*regexp.Regexp{navPattern, remarkPattern, selPattern, pbnPattern, eetPattern, performancePattern, reroutePattern} ) +var mu sync.Mutex + type BodyParser struct { + bodyMu sync.Mutex bodyPatterns map[string]config.BodyConfig } -// NewBodyParser initializes a BodyParser with the default body patterns. func NewBodyParser() *BodyParser { return &BodyParser{bodyPatterns: config.GetBodyPatterns()} } -// GetBodyPatterns returns the body patterns used by the parser. func (bp *BodyParser) GetBodyPatterns() map[string]config.BodyConfig { return bp.bodyPatterns } -// SetBodyPatterns sets the body patterns for the parser. func (bp *BodyParser) SetBodyPatterns(patterns map[string]config.BodyConfig) { bp.bodyPatterns = patterns } -// Parse attempts to parse the body text using the configured patterns. func (bp *BodyParser) Parse(body string) (string, interface{}, error) { + bp.bodyMu.Lock() + defer bp.bodyMu.Unlock() body = strings.TrimSpace(body) category := findCategory(body) if category == "" { @@ -61,14 +64,13 @@ func (bp *BodyParser) Parse(body string) (string, interface{}, error) { for _, p := range patternConfig.Patterns { if match := p.Expression.FindStringSubmatch(body); match != nil { data := extractData(match, p.Expression) - return createBodyData(data) + return bp.createBodyData(data) } } } return "", nil, fmt.Errorf("no matching pattern found for body: %s", body) } -// findCategory extracts the category from the body text using regex. func findCategory(body string) string { if match := categoryRegex.FindStringSubmatch(body); match != nil { for i, name := range categoryRegex.SubexpNames() { @@ -80,7 +82,6 @@ func findCategory(body string) string { return "" } -// extractData extracts named groups from the regex match and returns them as a map. func extractData(match []string, re *regexp.Regexp) map[string]string { data := make(map[string]string) for i, name := range re.SubexpNames() { @@ -91,8 +92,7 @@ func extractData(match []string, re *regexp.Regexp) map[string]string { return data } -// createBodyData creates the appropriate domain object based on the type of message. -func createBodyData(data map[string]string) (string, interface{}, error) { +func (bp *BodyParser) createBodyData(data map[string]string) (string, interface{}, error) { switch category := data["category"]; category { case "ARR": return category, &domain.ARR{ @@ -139,12 +139,13 @@ func createBodyData(data map[string]string) (string, interface{}, error) { Remarks: otherData["remark"], }, nil default: - return category, nil, fmt.Errorf("cann't parse : %s", category) + return category, nil, fmt.Errorf("cannot parse: %s", category) } } -// Parse parses the raw text message and returns a ParsedMessage. func Parse(rawText string) (*domain.ParsedMessage, error) { + mu.Lock() + defer mu.Unlock() message, err := ParseHeader(rawText) if err != nil { return nil, err @@ -163,7 +164,6 @@ func Parse(rawText string) (*domain.ParsedMessage, error) { return &message, nil } -// clean removes empty lines from a given text and extracts the body only. func clean(text string) string { cleanedText := emptyLineRemove.ReplaceAllString(text, "") cleanText := strings.ReplaceAll(cleanedText, "\n\n", "\n") @@ -177,11 +177,16 @@ func clean(text string) string { return "" } -// ParseHeader parses the header of the message and returns a ParsedMessage struct. func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { + log := utils.GetSugaredLogger() fullMessage = clean(fullMessage) lines := strings.Split(fullMessage, "\n") + if len(lines) < 3 { + log.Warnf("invalid message format: %s", fullMessage) + return domain.ParsedMessage{}, fmt.Errorf("invalid message format: %s", fullMessage) + } + _, messageID, dateTime, err := parseStartIndicator(lines[0]) if err != nil { return domain.ParsedMessage{}, err @@ -203,25 +208,24 @@ func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { }, nil } -// parseStartIndicator parses the start indicator line. func parseStartIndicator(line string) (string, string, string, error) { parts := strings.Fields(line) if len(parts) >= 3 && strings.HasPrefix(parts[0], StartIndicatorPrefix) { return parts[0], parts[1], parts[2], nil } + utils.GetSugaredLogger().Warnf("invalid start indicator line format: %s", line) return "", "", "", fmt.Errorf("invalid start indicator line format: %s", line) } -// parsePriorityAndPrimary parses the priority indicator and primary address line. func parsePriorityAndPrimary(line string) (string, string) { parts := strings.Fields(line) if len(parts) >= 2 { return parts[0], parts[1] } + utils.GetSugaredLogger().Warnf("invalid priority and primary address line format: %s", line) return "", "" } -// parseRemainingLines parses the remaining lines of the message. func parseRemainingLines(lines []string) ([]string, string, string, string) { var ( secondaryAddresses []string @@ -238,7 +242,6 @@ func parseRemainingLines(lines []string) ([]string, string, string, string) { } else { switch { case line == EndHeaderMarker: - // End header marker, do nothing case strings.HasPrefix(line, "."): originatorInfo := strings.Fields(line[1:]) if len(originatorInfo) >= 2 { @@ -266,16 +269,15 @@ func parseRemainingLines(lines []string) ([]string, string, string, string) { return secondaryAddresses, originator, originatorDateTime, bodyAndFooter.String() } -// getOriginator extracts originator details from a line of text. func getOriginator(line string) (string, string) { match := originator.FindStringSubmatch(line) if len(match) >= 3 { return match[1], match[2] } + utils.GetSugaredLogger().Warnf("invalid originator line format: %s", line) return "", "" } -// parseOther parses additional information from the message body. func parseOther(text string) map[string]string { data := make(map[string]string) for _, re := range otherPatterns { diff --git a/internal/parsers/aviation_parser_test.go b/internal/parsers/aviation_parser_test.go index 1ebc47c..1a7c484 100644 --- a/internal/parsers/aviation_parser_test.go +++ b/internal/parsers/aviation_parser_test.go @@ -10,39 +10,32 @@ import ( var _ = Describe("Aviation Parser", func() { Describe("ParseHeader", func() { - Context("with a real arr context", func() { + Context("with a real ARR context", func() { message := `ZCZC TMQ2530 141614 - - GG ZBTJZXZX - - 141614 ZSHCZTZX - (ARR-CES5470-ZBTJ-ZSHC1614) - NNNN` - It("get a clean body text", func() { + It("should get a clean body text", func() { body := clean(message) - expexted := `ZCZC TMQ2530 141614 + expected := `ZCZC TMQ2530 141614 GG ZBTJZXZX 141614 ZSHCZTZX (ARR-CES5470-ZBTJ-ZSHC1614)` - // fmt.Printf("\n%v \n%v\n", []byte(body), []byte(expexted)) - Expect(body).To(Equal(expexted)) + Expect(body).To(Equal(expected)) }) }) It("should parse the header correctly", func() { message := ` - ZCZC TAF6789 160530 +ZCZC TAF6789 160530 QU TSNZPCA . QU PEKUDCA TSNUOCA TSNZPCA TSNUFCA .TAF WSSS 160500Z 1606/1712 20010KT 9999 SCT018 - BECMG 1608/1610 24012KT 9999 SCT018 - TEMPO 1610/1612 4000 SHRA BKN012 - BECMG 1612/1614 18008KT 9999 SCT020 +BECMG 1608/1610 24012KT 9999 SCT018 +TEMPO 1610/1612 4000 SHRA BKN012 +BECMG 1612/1614 18008KT 9999 SCT020 BEGIN PART 02 @@ -54,13 +47,11 @@ ALTERNATE ROUTES ADVISED) NNNN` parsedHeader, err := ParseHeader(message) Expect(err).ToNot(HaveOccurred()) - // Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) Expect(parsedHeader.MessageID).To(Equal("TAF6789")) Expect(parsedHeader.DateTime).To(Equal("160530")) Expect(parsedHeader.PriorityIndicator).To(Equal("QU")) Expect(parsedHeader.PrimaryAddress).To(Equal("TSNZPCA")) Expect(parsedHeader.SecondaryAddresses).To(Equal([]string{"QU PEKUDCA TSNUOCA TSNZPCA TSNUFCA"})) - }) It("should parse the header correctly with originator information", func() { @@ -85,7 +76,7 @@ WE ACKNOWLEDGE THE RUNWAY CLOSURE. - CONTROL TOWER: - SIGN . . . . . . . . . . +SIGN . . . . . . . . . . BEGIN PART 02 @@ -97,7 +88,6 @@ ALL DEPARTURES/ARRIVALS EXPECTED TO BE DELAYED) NNNN` parsedHeader, err := ParseHeader(message) Expect(err).ToNot(HaveOccurred()) - // Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) Expect(parsedHeader.MessageID).To(Equal("NOTAM1122")) Expect(parsedHeader.DateTime).To(Equal("171000")) Expect(parsedHeader.PriorityIndicator).To(Equal("QU")) @@ -210,7 +200,6 @@ NNNN` Expect(fplMessage.DestinationAndTotalTime).To(Equal("ZBAA0153")) Expect(fplMessage.AlternateAirport).To(Equal("ZBYN")) Expect(fplMessage.OtherInfo).To(Equal("PBN/A1B2B3B4B5D1L1 NAV/ABAS REG/B6513 EET/ZBPE0112 SEL/KMAL PER/C RIF/FRT N640 ZBYN RMK/TCAS EQUIPPED")) - // Expect(fplMessage.SupplementaryInfo).To(Equal("RMK/TCAS EQUIPPED")) Expect(fplMessage.PBN).To(Equal("A1B2B3B4B5D1L1")) Expect(fplMessage.EstimatedElapsedTime).To(Equal("ZBPE0112")) Expect(fplMessage.SELCALCode).To(Equal("KMAL")) @@ -219,29 +208,21 @@ NNNN` Expect(fplMessage.Remarks).To(Equal("TCAS EQUIPPED")) }) }) + }) Describe("Parse whole real message", func() { - Context("with a real arr message", func() { + Context("with a real ARR message", func() { message := ` ZCZC TMQ2526 141605 - FF ZBTJZPZX - 141604 ZBACZQZX - (ARR-JAE7433/A0132-RKSI-ZBTJ1604) - - - - - - NNNN ` It("should parse the whole message correctly", func() { @@ -265,77 +246,5 @@ NNNN Expect(arrmsg.ArrivalTime).To(Equal("1604")) }) }) - - Context("real fpl", func() { - message := `ZCZC TMQ2544 141652 - - -FF ZBTJZXZX - - -141652 ZBTJZPZX - - -(FPL-JAE7433-IS - - --B744/H-SXIRPZJWY/S - - --ZBTJ1755 - - --K0926S0920 CG A326 VYK W80 HUR B339 GM A575 MANSA/K0919S0980 A575 - - - INTIK/K0917S0960 A575 UDA DCT BULAG A200 HATGA/K0900S1060 A308 - - - LARNA DCT RATKO A307 KUMOD R497 TODES B228 ZJ R22 KTL R30 SPB - - - B141 RANVA/N0485F360 UP863 DEREX UP739 KOLJA UN746 GORPI UZ80 - - - TILAV UL87 TADUV T173 GED GED2W - - --EDDF0948 EDDK - - --EET/ZMUB0100 UNKL0236 UNWW0332 UNNT0332 USRR0447 USHH0507 - - - USSS0535 UUYY0602 ULKK0634 ULWW0653 ULLL0720 EETT0748 EVRR0815 - - - ESAA0821 EPWW0848 EDUU0900 - - - REG/B2422 SEL/JLAD OPR/JADE CARGO DAT/S RVR/200 - - - NAV/RNAV1 RNAV5 RNP4 - - - RMK/AGCS EQUIPPED - - - ACARS EQUIPPED/TCAS EQUIPPED/FOREIGN PILOT - - - E/1148 P/TBN R/UV S/M J/LF D/1 15 C YELLOW - - - A/WHITE GREEN) - -NNNN -` - It("should parse the whole message correctly", func() { - parsedMessage, err := Parse(message) - Expect(err).ToNot(HaveOccurred()) - Expect(parsedMessage).ToNot(BeNil()) - }) - }) }) }) diff --git a/pkg/utils/log.go b/pkg/utils/log.go index 93eeaf6..92d0ad4 100644 --- a/pkg/utils/log.go +++ b/pkg/utils/log.go @@ -46,15 +46,13 @@ func load() { env := getEnv() fmt.Printf("Enviroment : %s\n", env) configFile := getConfigFile(env) - // fmt.Printf("Config File : %s\n", configFile) - // if err != nil { - // fmt.Printf("Error finding config file: %v\n", err) - // return - // } - // fmt.Printf("Loading config from file: %s\n", configFile) + config, err := loadConfig(configFile) if err != nil { fmt.Printf("Error loading config: %v\n", err) + //create a default logger + log, _ = zap.NewDevelopment() + sugar = log.Sugar() return }