windyboy
766925f7f9
The code changes in `sub.go` update the handling of NATS subscriptions. The `Subscribe` function now takes in a `config` parameter, allowing for more flexibility in configuring the subscription. Additionally, the `Subscribe` function now uses the `handler` and `marshaler` parameters to handle incoming messages and marshal/unmarshal data, respectively. These changes improve the modularity and extensibility of the code when working with NATS subscriptions.
71 lines
1.8 KiB
Go
71 lines
1.8 KiB
Go
package handlers
|
|
|
|
import (
|
|
"caatsm/internal/config"
|
|
"caatsm/internal/domain"
|
|
"caatsm/internal/parsers"
|
|
"caatsm/internal/repository"
|
|
"caatsm/pkg/utils"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
"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) 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
|
|
}
|