windyboy
dd75f0f705
The code changes in `aviation_parser.go` and `aviation_parser_test.go` enhance the CNL message parsing logic in the `BodyParser` module. This ensures accurate extraction of data from CNL messages, including the category, aircraft ID, departure airport, and destination airport. The improvements improve the overall functionality and accuracy of the code.
110 lines
2.9 KiB
Go
110 lines
2.9 KiB
Go
package nats
|
|
|
|
import (
|
|
"caatsm/internal/config"
|
|
"caatsm/internal/domain"
|
|
"caatsm/internal/parsers"
|
|
"caatsm/internal/repository"
|
|
"caatsm/pkg/utils"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
"github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
|
|
"github.com/ThreeDotsLabs/watermill/message"
|
|
nc "github.com/nats-io/nats.go"
|
|
)
|
|
|
|
type NatsHandler struct {
|
|
mu sync.Mutex
|
|
config *config.Config
|
|
hasuraRepo *repository.HasuraRepository
|
|
}
|
|
|
|
func NewNatsHandler(config *config.Config) *NatsHandler {
|
|
return &NatsHandler{config: config, hasuraRepo: repository.NewHasuraRepo(config.Hasura.Endpoint, config.Hasura.Secret)}
|
|
}
|
|
|
|
func (n *NatsHandler) Subscribe() {
|
|
marshaler := &PlainTextMarshaler{}
|
|
logger := watermill.NewStdLogger(false, false)
|
|
options := []nc.Option{
|
|
nc.RetryOnFailedConnect(true),
|
|
nc.Timeout(n.config.Timeouts.Server),
|
|
nc.ReconnectWait(n.config.Timeouts.ReconnectWait),
|
|
}
|
|
jsConfig := nats.JetStreamConfig{Disabled: true}
|
|
|
|
subscriber, err := nats.NewSubscriber(
|
|
nats.SubscriberConfig{
|
|
URL: n.config.Nats.URL,
|
|
CloseTimeout: n.config.Timeouts.Close,
|
|
AckWaitTimeout: n.config.Timeouts.AckWait,
|
|
NatsOptions: options,
|
|
Unmarshaler: marshaler,
|
|
JetStream: jsConfig,
|
|
},
|
|
logger,
|
|
)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
logger.Info("Subscribing to NATS topic", map[string]interface{}{"topic": n.config.Subscription.Topic})
|
|
|
|
defer subscriber.Close()
|
|
messages, err := subscriber.Subscribe(context.Background(), n.config.Subscription.Topic)
|
|
if err != nil {
|
|
logger.Error("Failed to subscribe to NATS topic", err, map[string]interface{}{"topic": n.config.Subscription.Topic})
|
|
return
|
|
}
|
|
for msg := range messages {
|
|
n.handleMessage(msg)
|
|
msg.Ack()
|
|
}
|
|
}
|
|
func (n *NatsHandler) handleMessage(msg *message.Message) error {
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
log := utils.GetSugaredLogger()
|
|
if msg.Payload == nil {
|
|
log.Error("empty message")
|
|
return fmt.Errorf("empty message")
|
|
}
|
|
payload := string(msg.Payload)
|
|
var parsed *domain.ParsedMessage
|
|
if parsed = parsers.Parse(payload); !parsed.Parsed {
|
|
log.Infof("not parsed: [%s] : {%s} \n", msg.UUID, msg.Payload)
|
|
} else {
|
|
log.Infof("parsed [%s]: %v\n", msg.UUID, parsed)
|
|
|
|
}
|
|
n.SaveMessage(parsed, msg.UUID)
|
|
return nil
|
|
}
|
|
|
|
func (n *NatsHandler) SaveMessage(parsed *domain.ParsedMessage, uuid string) {
|
|
logger := watermill.NewStdLogger(false, false)
|
|
if parsed != nil {
|
|
parsed.Uuid = uuid
|
|
if err := n.hasuraRepo.CreateNew(parsed); err != nil {
|
|
logger.Error("error inserting message", err, map[string]interface{}{"message": parsed})
|
|
}
|
|
}
|
|
}
|
|
|
|
type PlainTextMarshaler struct{}
|
|
|
|
func (m *PlainTextMarshaler) Marshal(topic string, msg nc.Msg) ([]byte, error) {
|
|
return msg.Data, nil
|
|
}
|
|
|
|
func (m *PlainTextMarshaler) Unmarshal(newMsg *nc.Msg) (*message.Message, error) {
|
|
if newMsg == nil {
|
|
return nil, errors.New("empty message")
|
|
}
|
|
msg := message.NewMessage(watermill.NewUUID(), newMsg.Data)
|
|
return msg, nil
|
|
}
|