core/oc-lib
5
0
Fork 0
Code Issues 3 Pull Requests Packages Projects Releases Wiki Activity

Discovery Nats automate

Browse Source
This commit is contained in:
mr 2024-08-22 15:38:27 +02:00
parent 8c9a983216
commit d89826db44
1 changed files with 3 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
tools/nats_caller.go
View File

@ -4,7 +4,6 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"strings" "strings"
"sync"
"cloud.o-forge.io/core/oc-lib/logs" "cloud.o-forge.io/core/oc-lib/logs"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
@ -73,17 +72,12 @@ func (o *natsCaller) DiscoveryNATS(name string, model NATSObjectInterface) error
if err != nil { if err != nil {
return errors.New("Could not connect to NATS") return errors.New("Could not connect to NATS")
} }
defer nc.Close() go o.listenForChange(nc, model, DISCOVERY.GenerateKey("ask"))
wg := &sync.WaitGroup{} go o.listenForChange(nc, model, DISCOVERY.GenerateKey(name))
wg.Add(2)
go o.listenForChange(nc, model, DISCOVERY.GenerateKey("ask"), wg)
go o.listenForChange(nc, model, DISCOVERY.GenerateKey(name), wg)
wg.Wait()
return nil return nil
} }
func (o *natsCaller) listenForChange(nc *nats.Conn, model NATSObjectInterface, chanName string, wg *sync.WaitGroup) { func (o *natsCaller) listenForChange(nc *nats.Conn, model NATSObjectInterface, chanName string) {
defer wg.Done()
api := API{} api := API{}
logger := logs.GetLogger() logger := logs.GetLogger()
ch := make(chan *nats.Msg, 64) ch := make(chan *nats.Msg, 64)
@ -93,7 +87,6 @@ func (o *natsCaller) listenForChange(nc *nats.Conn, model NATSObjectInterface, c
return return
} }
defer subs.Unsubscribe() defer subs.Unsubscribe()
for msg := range ch { for msg := range ch {
logger.Info().Msg("Received message from NATS : " + string(msg.Data)) logger.Info().Msg("Received message from NATS : " + string(msg.Data))
m := model.Serialize() m := model.Serialize()