Configuration NATS

config/conf.go

@@ -9,14 +9,16 @@ import "sync"
 // ===================================================
 
 type Config struct {
-	NATSUrl       string
+	NATSUrl        string
-	MongoUrl      string
+	MongoUrl       string
-	MongoDatabase string
+	MongoDatabase  string
-	Host          string
+	Host           string
-	Port          string
+	Port           string
-	LokiUrl       string
+	LokiUrl        string
-	LogLevel      string
+	LogLevel       string
-	Whitelist     bool
+	Whitelist      bool
+	PrivateKeyPath string
+	PublicKeyPath  string
 }
 
 func (c Config) GetUrl() string {

tools/api.go

@@ -61,7 +61,7 @@ func (s State) String() string {
 type API struct{}
 
 func (a *API) Discovered(infos []*beego.ControllerInfo) {
-	respondToDiscovery := func(m map[string]interface{}) {
+	respondToDiscovery := func(m map[string]string) {
 		if len(m) == 0 {
 			a.SubscribeRouter(infos)
 		}
@@ -91,8 +91,10 @@ func (a *API) GetState() (State, int, error) {
 	return ALIVE, 200, nil // If everything is up, return alive
 }
 
-func (a *API) ListenRouter(exec func(msg map[string]interface{})) {
+func (a *API) ListenRouter(exec func(msg map[string]string)) {
-	go NewNATSCaller().ListenNats(DISCOVERY.GenerateKey("api"), exec)
+	go NewNATSCaller().ListenNats(DISCOVERY.GenerateKey(), map[NATSMethod]func(msg map[string]string){
+		DISCOVERY: exec,
+	})
 }
 
 func (a *API) SubscribeRouter(infos []*beego.ControllerInfo) {

tools/nats_caller.go

@@ -2,27 +2,39 @@ package tools
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"cloud.o-forge.io/core/oc-lib/config"
 	"cloud.o-forge.io/core/oc-lib/logs"
 	"github.com/nats-io/nats.go"
+	"github.com/rs/zerolog"
 )
 
 // NATS Method Enum defines the different methods that can be used to interact with the NATS server
 type NATSMethod int
 
+var meths = []string{"remove execution", "create execution", "discovery",
+	"workflow event", "peer discovery", "peer detection"}
+
 const (
-	REMOVE NATSMethod = iota
+	REMOVE_EXECUTION NATSMethod = iota
-	CREATE
+	CREATE_EXECTUTION
 	DISCOVERY
 	WORKFLOW_EVENT
+	PEER_DISCOVERY
+	PEER_DETECTION
 )
 
+func (n NATSMethod) String() string {
+	return meths[n]
+}
+
 // NameToMethod returns the NATSMethod enum value from a string
 func NameToMethod(name string) NATSMethod {
-	for _, v := range [...]NATSMethod{REMOVE, CREATE} {
+	for _, v := range [...]NATSMethod{REMOVE_EXECUTION, CREATE_EXECTUTION, DISCOVERY, WORKFLOW_EVENT, PEER_DISCOVERY, PEER_DETECTION} {
 		if strings.Contains(strings.ToLower(v.String()), strings.ToLower(name)) {
 			return v
 		}
@@ -31,13 +43,8 @@ func NameToMethod(name string) NATSMethod {
 }
 
 // GenerateKey generates a key for the NATSMethod usefull for standard key based on data name & method
-func (d NATSMethod) GenerateKey(name string) string {
+func (d NATSMethod) GenerateKey() string {
-	return name + "_" + d.String()
+	return strings.ReplaceAll(d.String(), " ", "_")
-}
-
-// String returns the string of the enum
-func (d NATSMethod) String() string {
-	return [...]string{"remove", "create", "discovery"}[d]
 }
 
 type natsCaller struct{}
@@ -49,7 +56,7 @@ func NewNATSCaller() *natsCaller {
 
 // on workflows' scheduling. Messages must contain
 // workflow execution ID, to allow retrieval of execution infos
-func (s *natsCaller) ListenNats(chanName string, exec func(msg map[string]interface{})) {
+func (s *natsCaller) ListenNats(chanName string, execs map[NATSMethod]func(map[string]string)) {
 	log := logs.GetLogger()
 	if config.GetConfig().NATSUrl == "" {
 		log.Error().Msg(" -> NATS_SERVER is not set")
@@ -58,21 +65,17 @@ func (s *natsCaller) ListenNats(chanName string, exec func(msg map[string]interf
 	for {
 		nc, err := nats.Connect(config.GetConfig().NATSUrl)
 		if err != nil {
+			log.Error().Msg("Could not connect to NATS")
 			time.Sleep(1 * time.Minute)
 			continue
 		}
-		ch := make(chan *nats.Msg, 64)
+		defer nc.Close()
-		subs, err := nc.ChanSubscribe(chanName, ch)
+		var wg sync.WaitGroup
-		if err != nil {
+		wg.Add(len(execs))
-			log.Error().Msg("Error listening to NATS : " + err.Error())
+		for k, v := range execs {
-		}
+			go s.listenForChange(log, nc, k, v, &wg)
-		defer subs.Unsubscribe()
-
-		for msg := range ch {
-			map_mess := map[string]interface{}{}
-			json.Unmarshal(msg.Data, &map_mess)
-			exec(map_mess)
 		}
+		wg.Wait()
 		break
 	}
 }
@@ -93,7 +96,7 @@ func (o *natsCaller) SetNATSPub(dataName string, method NATSMethod, data interfa
 		if err != nil {
 			return " -> " + err.Error()
 		}
-		err = nc.Publish(method.GenerateKey(dataName), js) // Publish the message on the NATS server with a channel name based on the data name (or whatever start) and the method
+		err = nc.Publish(method.GenerateKey(), js) // Publish the message on the NATS server with a channel name based on the data name (or whatever start) and the method
 		if err != nil {
 			time.Sleep(1 * time.Minute)
 			continue
@@ -102,3 +105,25 @@ func (o *natsCaller) SetNATSPub(dataName string, method NATSMethod, data interfa
 	}
 	return ""
 }
+
+// Goroutine listening to a NATS server for updates
+// on workflows' scheduling. Messages must contain
+// workflow execution ID, to allow retrieval of execution infos
+func (o *natsCaller) listenForChange(logger zerolog.Logger, nc *nats.Conn, natsTools NATSMethod,
+	function func(map[string]string), wg *sync.WaitGroup) {
+	defer wg.Done()
+	ch := make(chan *nats.Msg, 64)
+	logger.Info().Msg("Listening to " + natsTools.GenerateKey())
+	subs, err := nc.ChanSubscribe(natsTools.GenerateKey(), ch)
+	if err != nil {
+		logger.Error().Msg("Error listening to NATS : " + err.Error())
+	}
+	defer subs.Unsubscribe()
+
+	for msg := range ch {
+		map_mess := map[string]string{}
+		json.Unmarshal(msg.Data, &map_mess)
+		fmt.Println("Catching " + natsTools.String() + " workflow... " + map_mess["id"])
+		function(map_mess)
+	}
+}