From 53f03bbc0a5715636974342c29b07293a8831102 Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 22 Jul 2024 16:23:15 +0800 Subject: [PATCH] refactor: Update config.dev.toml and sub.go for Hasura integration The code changes in config.dev.toml and sub.go update the configuration and message handling for integrating with Hasura. The config.dev.toml file now includes a new section for Hasura configuration, specifying the endpoint and secret. In sub.go, the UUID field is set to the UUID of the received message, ensuring proper tracking of parsed messages. This refactor enables seamless integration with Hasura and improves the functionality of the code. --- configs/config.dev.toml | 6 +- go.sum | 10 ++- internal/config/config.go | 6 ++ internal/config/config_test.go | 7 ++ internal/domain/aviation.go | 3 +- internal/nats/sub.go | 1 + internal/parsers/aviation_parser.go | 20 +++++- internal/parsers/aviation_parser_test.go | 4 +- internal/repository/hasura.go | 83 ++++++++++++++++++++++++ 9 files changed, 131 insertions(+), 9 deletions(-) create mode 100644 internal/repository/hasura.go diff --git a/configs/config.dev.toml b/configs/config.dev.toml index daa7635..3bcdbac 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -11,4 +11,8 @@ queue = "tele-queue" server = "5s" reconnect_wait = "5s" close = "10s" -ack_wait = "5s" \ No newline at end of file +ack_wait = "5s" + +[hasura] +endpoint = "http://localhost:8080/v1/graphql" +secret = "aviation-test" \ No newline at end of file diff --git a/go.sum b/go.sum index 533a7fd..6a6f9d1 100644 --- a/go.sum +++ b/go.sum @@ -19,14 +19,16 @@ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6 h1:k7nVchz72niMH6YLQNvHSdIE7iqsQxK1P41mySCvssg= github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw= github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4= -github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo= github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= +github.com/hasura/go-graphql-client v0.12.2 h1:cYeQK/CELtvFy2jvik4kG0b5UMGngQRYWTTXQkbGHDo= +github.com/hasura/go-graphql-client v0.12.2/go.mod h1:17qYcHgGSensF/wMAHKUhtMYaRZwZa3TyD7biqH9L3k= github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4= github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -97,6 +99,8 @@ golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjs golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k= golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac= golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= +golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs= +golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI= golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk= @@ -115,3 +119,5 @@ gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYs gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +nhooyr.io/websocket v1.8.11 h1:f/qXNc2/3DpoSZkHt1DQu6rj4zGC8JmkkLkWss0MgN0= +nhooyr.io/websocket v1.8.11/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/internal/config/config.go b/internal/config/config.go index eeea758..224d778 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -20,6 +20,7 @@ type Config struct { Nats NatsConfig Subscription SubscriptionConfig Timeouts TimeoutsConfig + Hasura HasuraConfig } type NatsConfig struct { @@ -50,6 +51,11 @@ type PatternConfig struct { Expression *regexp.Regexp } +type HasuraConfig struct { + Endpoint string + Secret string +} + // LoggerConfig represents the configuration for the logger. type LoggerConfig struct { ZapConfig zap.Config `json:"zapConfig"` diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 033a90c..08af0ec 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -38,6 +38,10 @@ server_timeout = "30s" reconnect_wait = "10s" close_timeout = "10s" ack_wait_timeout = "5s" + +[hasura] +endpoint = "http://localhost:8080/v1/graphql" +secret = "aviation-test" ` tmpFile, err := os.CreateTemp("", "config.*.toml") Expect(err).NotTo(HaveOccurred()) @@ -68,6 +72,9 @@ ack_wait_timeout = "5s" Expect(cfg.Nats.Client).To(Equal("test-client")) Expect(cfg.Nats.URL).To(Equal("nats://localhost:4222")) Expect(cfg.Subscription.Topic).To(Equal("example-topic")) + Expect(cfg.Subscription.QueueGroup).To(Equal("example-group")) + Expect(cfg.Hasura.Endpoint).To(Equal("http://localhost:8080/v1/graphql")) + Expect(cfg.Hasura.Secret).To(Equal("aviation-test")) }) }) diff --git a/internal/domain/aviation.go b/internal/domain/aviation.go index de88302..2b3e50b 100644 --- a/internal/domain/aviation.go +++ b/internal/domain/aviation.go @@ -86,7 +86,8 @@ NeedDispatch: false. // ParsedMessage holds the parsed data from an aviation message type ParsedMessage struct { - StartIndicator string `json:"startIndicator"` // 电报开始标识: The start of the message indicator (e.g., 'ZCZC'). + // StartIndicator string `json:"startIndicator"` // 电报开始标识: The start of the message indicator (e.g., 'ZCZC'). + Uuid string `json:"uuid"` MessageID string `json:"messageId"` // 信息ID: The message ID (e.g., 'TMQ1324'). DateTime string `json:"dateTime"` // 日期时间: The date and time of the message (e.g., '150631'). PriorityIndicator string `json:"priorityIndicator"` // 优先级标识: The priority level of the message (e.g., 'FF'). diff --git a/internal/nats/sub.go b/internal/nats/sub.go index 0e8236d..b433b6c 100644 --- a/internal/nats/sub.go +++ b/internal/nats/sub.go @@ -67,6 +67,7 @@ func (n *NatsHandler) handleMessage(msg *message.Message) error { payload := string(msg.Payload) if parsed, err := parsers.Parse(payload); err != nil { // log.Error("error parsing message", err, map[string]interface{}{"payload": payload}) + parsed.Uuid = msg.UUID fmt.Print("error parsing message", err) return err } else { diff --git a/internal/parsers/aviation_parser.go b/internal/parsers/aviation_parser.go index dd1e1ec..727814f 100644 --- a/internal/parsers/aviation_parser.go +++ b/internal/parsers/aviation_parser.go @@ -149,19 +149,33 @@ func createBodyData(data map[string]string) (interface{}, error) { } } +// Parse parses the raw text message and returns a ParsedMessage. // Parse parses the raw text message and returns a ParsedMessage. func Parse(rawText string) (*domain.ParsedMessage, error) { + // Parse the header of the message message, err := ParseHeader(rawText) if err != nil { return nil, err } + + // Initialize a new body parser bodyParser := NewBodyParser() + + // Parse the body and footer of the message bodyData, err := bodyParser.Parse(message.BodyAndFooter) if err != nil { - return nil, err + // Return the message with the parsed header and the error + // message.ParsedAt = time.Now() + return &message, err } + + // Set the parsed time to the current time message.ParsedAt = time.Now() + + // Assign the parsed body data to the message message.BodyData = bodyData + + // Return the fully parsed message return &message, nil } @@ -188,7 +202,7 @@ func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { // fullMessage = strings.TrimSpace(fullMessage) lines := strings.Split(fullMessage, "\n") - startIndicator, messageID, dateTime, err := parseStartIndicator(lines[0]) + _, messageID, dateTime, err := parseStartIndicator(lines[0]) if err != nil { return domain.ParsedMessage{}, err } @@ -202,7 +216,7 @@ func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { secondaryAddresses, originator, originatorDateTime, bodyAndFooter := parseRemainingLines(lines[2:]) return domain.ParsedMessage{ - StartIndicator: startIndicator, + // StartIndicator: startIndicator, MessageID: messageID, DateTime: dateTime, PriorityIndicator: priorityIndicator, diff --git a/internal/parsers/aviation_parser_test.go b/internal/parsers/aviation_parser_test.go index 07c746f..d17cce0 100644 --- a/internal/parsers/aviation_parser_test.go +++ b/internal/parsers/aviation_parser_test.go @@ -54,7 +54,7 @@ ALTERNATE ROUTES ADVISED) NNNN` parsedHeader, err := ParseHeader(message) Expect(err).ToNot(HaveOccurred()) - Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) + // Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) Expect(parsedHeader.MessageID).To(Equal("TAF6789")) Expect(parsedHeader.DateTime).To(Equal("160530")) Expect(parsedHeader.PriorityIndicator).To(Equal("QU")) @@ -97,7 +97,7 @@ ALL DEPARTURES/ARRIVALS EXPECTED TO BE DELAYED) NNNN` parsedHeader, err := ParseHeader(message) Expect(err).ToNot(HaveOccurred()) - Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) + // Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) Expect(parsedHeader.MessageID).To(Equal("NOTAM1122")) Expect(parsedHeader.DateTime).To(Equal("171000")) Expect(parsedHeader.PriorityIndicator).To(Equal("QU")) diff --git a/internal/repository/hasura.go b/internal/repository/hasura.go new file mode 100644 index 0000000..a812bf8 --- /dev/null +++ b/internal/repository/hasura.go @@ -0,0 +1,83 @@ +package repository + +import ( + "context" + "fmt" + "os" + + "caatsm/internal/domain" + + "github.com/hasura/go-graphql-client" + "golang.org/x/oauth2" +) + +type HasuraRepository struct { + hasuraClient *graphql.Client +} + +// NewHasuraClient creates a new HasuraClient +func NewHasuraRepo(endpoint, secret string) *HasuraRepository { + src := oauth2.StaticTokenSource( + &oauth2.Token{AccessToken: os.Getenv("GRAPHQL_TOKEN")}, + ) + httpClient := oauth2.NewClient(context.Background(), src) + return &HasuraRepository{ + hasuraClient: graphql.NewClient(endpoint, httpClient), + } +} + +var mutation struct { + InsertParsedMessages struct { + Returning []domain.ParsedMessage `json:"returning"` + } ` + graphql:"insert_parsed_messages(objects: { + startIndicator: $startIndicator, + messageId: $messageId, + dateTime: $dateTime, + priorityIndicator: $priorityIndicator, + primaryAddress: $primaryAddress, + secondaryAddresses: $secondaryAddresses, + originator: $originator, + originatorDateTime: $originatorDateTime, + category: $category, + bodyAndFooter: $bodyAndFooter, + bodyData: $bodyData, + receivedAt: $receivedAt, + parsedAt: $parsedAt, + dispatchedAt: $dispatchedAt, + needDispatch: $needDispatch + })"` +} + +// InsertParsedMessage inserts a new ParsedMessage into the Hasura GraphQL API +func (hr *HasuraRepository) InsertParsedMessage(pm domain.ParsedMessage) error { + + variables := map[string]interface{}{ + // "startIndicator": graphql.String(pm.StartIndicator), + "messageId": graphql.String(pm.MessageID), + "dateTime": graphql.String(pm.DateTime), + "priorityIndicator": graphql.String(pm.PriorityIndicator), + "primaryAddress": graphql.String(pm.PrimaryAddress), + "secondaryAddresses": pm.SecondaryAddresses, + "originator": graphql.String(pm.Originator), + "originatorDateTime": graphql.String(pm.OriginatorDateTime), + "category": graphql.String(pm.Category), + "bodyAndFooter": graphql.String(pm.BodyAndFooter), + "bodyData": pm.BodyData, + "receivedAt": pm.ReceivedAt, + "parsedAt": pm.ParsedAt, + "dispatchedAt": pm.DispatchedAt, + "needDispatch": pm.NeedDispatch, + } + + ctx := context.Background() + err := hr.hasuraClient.Mutate(ctx, &mutation, variables) + if err != nil { + return err + } + + for _, returnedMessage := range mutation.InsertParsedMessages.Returning { + fmt.Printf("Inserted ParsedMessage: %+v\n", returnedMessage) + } + return nil +}