diff --git a/cmd/main/main.go b/cmd/main/main.go index 056db09..3c10d16 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -2,9 +2,8 @@ package main import ( "caatsm/internal/config" + "caatsm/internal/nats" "caatsm/pkg/utils" - - "github.com/sirupsen/logrus" ) func main() { @@ -19,8 +18,6 @@ func main() { utils.Logger.Info("Loaded configuration successfully") - // Example usage - utils.Logger.WithFields(logrus.Fields{ - "url": cfg.Nats.URL, - }).Info("NATS configuration") + subscribe := nats.NewNatsHandler(cfg) + subscribe.Subscribe() } diff --git a/internal/config/config.go b/internal/config/config.go index 838377d..2c39207 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -5,6 +5,7 @@ import ( "os" "regexp" "strings" + "time" "github.com/spf13/viper" ) @@ -29,10 +30,10 @@ type SubscriptionConfig struct { } type TimeoutsConfig struct { - ServerTimeout string `mapstructure:"server_timeout"` - ReconnectWait string `mapstructure:"reconnect_wait"` - CloseTimeout string `mapstructure:"close_timeout"` - AckWaitTimeout string `mapstructure:"ack_wait_timeout"` + ServerTimeout time.Duration `mapstructure:"server_timeout"` + ReconnectWait time.Duration `mapstructure:"reconnect_wait"` + CloseTimeout time.Duration `mapstructure:"close_timeout"` + AckWaitTimeout time.Duration `mapstructure:"ack_wait_timeout"` } type BodyConfig struct { diff --git a/internal/nats/sub.go b/internal/nats/sub.go index c8654eb..6bf78f3 100644 --- a/internal/nats/sub.go +++ b/internal/nats/sub.go @@ -1,5 +1,90 @@ package nats -func Subscribe() { - // Placeholder for the NATS subscription logic. +import ( + "caatsm/internal/config" + "caatsm/internal/parsers" + "caatsm/pkg/utils" + "context" + "errors" + "fmt" + + "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 { + config *config.Config +} + +func NewNatsHandler(config *config.Config) *NatsHandler { + return &NatsHandler{config: config} +} + +func (n *NatsHandler) Subscribe() { + marshaler := &PlainTextMarshaler{} + logger := watermill.NewStdLogger(false, false) + options := []nc.Option{ + nc.RetryOnFailedConnect(true), + nc.Timeout(n.config.Timeouts.ServerTimeout), + 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.CloseTimeout, + AckWaitTimeout: n.config.Timeouts.AckWaitTimeout, + 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 { + log := utils.Logger + if msg.Payload == nil { + log.Error("empty message") + return fmt.Errorf("empty message") + } + payload := string(msg.Payload) + if parsed, err := parsers.Parse(payload); err != nil { + log.Error("error parsing message", err, map[string]interface{}{"payload": payload}) + return err + } else { + log.Info("message ", map[string]interface{}{"message": parsed}) + } + return nil +} + +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 }