This commit is contained in:
aespin 2024-10-02 15:36:58 +02:00
parent 49fb704bb1
commit 04ebecd647
3 changed files with 52 additions and 11 deletions

View File

@ -1,12 +1,16 @@
package app package app
import ( import (
"bytes"
"context" "context"
"encoding/json"
"fmt" "fmt"
"os" "os"
"os/signal" "os/signal"
"syscall" "syscall"
cml04eventer "git.espin.casa/albert/cml04-eventer"
"git.espin.casa/albert/cml04-falcon-system/internal/publisher"
"git.espin.casa/albert/cml04-falcon-system/internal/storage" "git.espin.casa/albert/cml04-falcon-system/internal/storage"
"git.espin.casa/albert/logger" "git.espin.casa/albert/logger"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
@ -56,13 +60,15 @@ func Run(cmd *cobra.Command, args []string) {
return return
} }
defer nc.Close() defer nc.Close()
// create publisher
pub := publisher.New(nc)
// create context // create context
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()
// NATS subscribe subject // NATS subscribe subject
nc.Subscribe("barcode.new", func(msg *nats.Msg) { nc.Subscribe("barcode.new", func(msg *nats.Msg) {
go func(log logger.LoggerAdapter) { go func(log logger.LoggerAdapter) {
if err := ProcessBarcodeMessage(ctx, msg, storage, log); err != nil { if err := ProcessNewBarcodeMessage(ctx, msg, storage, pub, log); err != nil {
log.Error("process new barcode message failed", err, logger.LogFields{}) log.Error("process new barcode message failed", err, logger.LogFields{})
return return
} }
@ -70,7 +76,7 @@ func Run(cmd *cobra.Command, args []string) {
}) })
nc.Subscribe("bundle.new", func(msg *nats.Msg) { nc.Subscribe("bundle.new", func(msg *nats.Msg) {
go func(log logger.LoggerAdapter) { go func(log logger.LoggerAdapter) {
if err := ProcessBundleMessage(ctx, msg, storage, log); err != nil { if err := ProcessNewBundleMessage(ctx, msg, storage, log); err != nil {
log.Error("process new bundle message failed", err, logger.LogFields{}) log.Error("process new bundle message failed", err, logger.LogFields{})
return return
} }
@ -85,11 +91,35 @@ func Run(cmd *cobra.Command, args []string) {
})) }))
} }
func ProcessBarcodeMessage(ctx context.Context, msg *nats.Msg, storage storage.Storager, log logger.LoggerAdapter) error { func ProcessNewBarcodeMessage(ctx context.Context, msg *nats.Msg, storage storage.Storager, pub publisher.Publisher, log logger.LoggerAdapter) error {
// event holder
event := &cml04eventer.Event{}
// create new buffer based on message data
buff := bytes.NewBuffer(msg.Data)
// decode message
if err := json.NewDecoder(buff).Decode(event); err != nil {
return err
}
// info banner
log.Info("received event bus message", logger.LogFields{
"event_id": event.EventID,
"event_topic": event.EventTopic,
})
// barcode
barcode := string(event.EventData)
// store barcode
if err := storage.ConfirmBundle(ctx, barcode); err != nil {
return err
}
// send bundle confirmation
if err := pub.ComfirmedBundle(ctx, barcode); err != nil {
return err
}
return nil return nil
} }
func ProcessBundleMessage(ctx context.Context, msg *nats.Msg, storage storage.Storager, log logger.LoggerAdapter) error { func ProcessNewBundleMessage(ctx context.Context, msg *nats.Msg, storage storage.Storager, log logger.LoggerAdapter) error {
return nil return nil
} }

View File

@ -4,19 +4,16 @@ Copyright © 2024 NAME HERE <EMAIL ADDRESS>
package cmd package cmd
import ( import (
"fmt" "git.espin.casa/albert/cml04-falcon-system/bundle/app"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
// bundleCmd represents the bundle command // bundleCmd represents the bundle command
var bundleCmd = &cobra.Command{ var bundleCmd = &cobra.Command{
Use: "bundle", Use: "bundle",
Short: "bundle service manager.", Short: "Bundle service manager.",
Long: "", Long: "Bundle service manager. Register, calculate values for bundles (production).",
Run: func(cmd *cobra.Command, args []string) { Run: app.Run,
fmt.Println("bundle called")
},
} }
func init() { func init() {

View File

@ -11,12 +11,26 @@ import (
type Publisher interface { type Publisher interface {
NewBarcode(ctx context.Context, barcode string) error NewBarcode(ctx context.Context, barcode string) error
ComfirmedBundle(ctx context.Context, barcode string) error
} }
type publisher struct { type publisher struct {
nc *nats.Conn nc *nats.Conn
} }
func (e *publisher) ComfirmedBundle(ctx context.Context, barcode string) error {
// create event message
evt := cml04eventer.NewEvent("falcon-handset", "bundle.confirmed", nil, []byte(barcode))
// create buffer message data
buf := bytes.Buffer{}
// encode message
if err := json.NewEncoder(&buf).Encode(evt); err != nil {
return err
}
// publish event message
return e.nc.Publish("bundle.confirmed", buf.Bytes())
}
func (e *publisher) NewBarcode(ctx context.Context, barcode string) error { func (e *publisher) NewBarcode(ctx context.Context, barcode string) error {
// create event message // create event message
evt := cml04eventer.NewEvent("falcon-handset", "barcode.new", nil, []byte(barcode)) evt := cml04eventer.NewEvent("falcon-handset", "barcode.new", nil, []byte(barcode))