Adapt With Close

infrastructure/nats.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"slices"
+	"sync"
 
 	oclib "cloud.o-forge.io/core/oc-lib"
 	"cloud.o-forge.io/core/oc-lib/config"
@@ -15,12 +16,15 @@ var ressourceCols = []oclib.LibDataEnum{
 	oclib.LibDataEnum(oclib.PEER),
 }
 
+var SearchMu sync.RWMutex
 var SearchStream = map[string]chan *peer.Peer{}
 
 func EmitNATS(user string, groups []string, message tools.PropalgationMessage) {
 	b, _ := json.Marshal(message)
 	if message.Action == tools.PB_SEARCH {
+		SearchMu.Lock()
 		SearchStream[user] = make(chan *peer.Peer, 128)
+		SearchMu.Unlock()
 	}
 	tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
 		FromApp:  "oc-peer",
@@ -53,7 +57,7 @@ func EmitNATS(user string, groups []string, message tools.PropalgationMessage) {
 func ListenNATS() {
 	tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
 		tools.CREATE_RESOURCE: func(resp tools.NATSResponse) {
-			if resp.FromApp == config.GetAppName() || resp.Datatype != tools.PEER {
+			if resp.FromApp == config.GetAppName() || !slices.Contains(ressourceCols, oclib.LibDataEnum(resp.Datatype)) {
 				return
 			}
 			p := &peer.Peer{}
@@ -96,7 +100,9 @@ func ListenNATS() {
 			err := json.Unmarshal(resp.Payload, p)
 			if err == nil {
 				fmt.Println("ADD in SEARCH STREAM")
+				SearchMu.Lock()
 				SearchStream[resp.User] <- p // TODO when do we update it in our catalog ?
+				SearchMu.Unlock()
 			}
 		},
 	})