refactor: Update Nats handler and Hasura repository initialization
This commit is contained in:
+3
-1
@@ -38,12 +38,14 @@ func setupApp() *cli.App {
|
|||||||
Flags: []cli.Flag{
|
Flags: []cli.Flag{
|
||||||
&cli.StringFlag{
|
&cli.StringFlag{
|
||||||
Name: "nats",
|
Name: "nats",
|
||||||
|
Aliases: []string{"n"},
|
||||||
Usage: "Nats server address",
|
Usage: "Nats server address",
|
||||||
Value: "nats://localhost:4222",
|
Value: "nats://localhost:4222",
|
||||||
EnvVars: []string{"NATS_SERVER"},
|
EnvVars: []string{"NATS_SERVER"},
|
||||||
},
|
},
|
||||||
&cli.StringFlag{
|
&cli.StringFlag{
|
||||||
Name: "topic",
|
Name: "topic",
|
||||||
|
Aliases: []string{"t"},
|
||||||
Usage: "Nats topic to listen to",
|
Usage: "Nats topic to listen to",
|
||||||
Value: "Telegram.Serial",
|
Value: "Telegram.Serial",
|
||||||
EnvVars: []string{"NATS_SUBJECT"},
|
EnvVars: []string{"NATS_SUBJECT"},
|
||||||
@@ -81,7 +83,7 @@ func executeListen(c *cli.Context) error {
|
|||||||
fmt.Println("Loaded configuration successfully")
|
fmt.Println("Loaded configuration successfully")
|
||||||
log := utils.GetLogger()
|
log := utils.GetLogger()
|
||||||
log.Info("Starting nats subscriber")
|
log.Info("Starting nats subscriber")
|
||||||
handler := handlers.NewNatsHandler(cfg)
|
handler := handlers.New(cfg)
|
||||||
nats.Subscribe(cfg, &handlers.PlainTextMarshaler{}, handler)
|
nats.Subscribe(cfg, &handlers.PlainTextMarshaler{}, handler)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -21,8 +21,11 @@ type MessageHandler struct {
|
|||||||
hasuraRepo *repository.HasuraRepository
|
hasuraRepo *repository.HasuraRepository
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewNatsHandler(config *config.Config) *MessageHandler {
|
func New(config *config.Config) *MessageHandler {
|
||||||
return &MessageHandler{config: config, hasuraRepo: repository.NewHasuraRepo(config.Hasura.Endpoint, config.Hasura.Secret)}
|
return &MessageHandler{
|
||||||
|
config: config,
|
||||||
|
hasuraRepo: repository.New(config.Hasura.Endpoint, config.Hasura.Secret),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (n *MessageHandler) HandleMessage(msg *message.Message) error {
|
func (n *MessageHandler) HandleMessage(msg *message.Message) error {
|
||||||
@@ -39,7 +42,6 @@ func (n *MessageHandler) HandleMessage(msg *message.Message) error {
|
|||||||
log.Infof("not parsed: [%s] : {%s} \n", msg.UUID, msg.Payload)
|
log.Infof("not parsed: [%s] : {%s} \n", msg.UUID, msg.Payload)
|
||||||
} else {
|
} else {
|
||||||
log.Infof("parsed [%s]: %v\n", msg.UUID, parsed)
|
log.Infof("parsed [%s]: %v\n", msg.UUID, parsed)
|
||||||
|
|
||||||
}
|
}
|
||||||
n.SaveMessage(parsed, msg.UUID)
|
n.SaveMessage(parsed, msg.UUID)
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -17,8 +17,8 @@ type HasuraRepository struct {
|
|||||||
client graphql.Client
|
client graphql.Client
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewHasuraRepo creates a new HasuraRepository
|
// New creates a new HasuraRepository
|
||||||
func NewHasuraRepo(endpoint, secret string) *HasuraRepository {
|
func New(endpoint, secret string) *HasuraRepository {
|
||||||
src := oauth2.StaticTokenSource(
|
src := oauth2.StaticTokenSource(
|
||||||
&oauth2.Token{AccessToken: os.Getenv("GRAPHQL_TOKEN")},
|
&oauth2.Token{AccessToken: os.Getenv("GRAPHQL_TOKEN")},
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user