package infrastructure import ( "encoding/json" "fmt" oclib "cloud.o-forge.io/core/oc-lib" "cloud.o-forge.io/core/oc-lib/models/resources" "cloud.o-forge.io/core/oc-lib/tools" ) var SearchStream = map[string]chan resources.ResourceInterface{} func EmitNATS(user string, message tools.PropalgationMessage) { b, _ := json.Marshal(message) if message.Action == tools.PB_SEARCH { SearchStream[user] = make(chan resources.ResourceInterface, 128) } tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{ FromApp: "oc-catalog", Datatype: -1, Method: int(tools.PROPALGATION_EVENT), Payload: b, }) } func ListenNATS() { tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){ tools.CATALOG_SEARCH_EVENT: func(resp tools.NATSResponse) { p, err := resources.ToResource(int(resp.Datatype), resp.Payload) if err == nil { SearchStream[resp.User] <- p } }, tools.CREATE_RESOURCE: func(resp tools.NATSResponse) { p := map[string]interface{}{} err := json.Unmarshal(resp.Payload, &p) if err == nil { access := oclib.NewRequestAdmin(oclib.LibDataEnum(resp.Datatype), nil) if data := access.LoadOne(fmt.Sprintf("%v", p["id"])); data.Data != nil { access.UpdateOne(p, fmt.Sprintf("%v", p["id"])) } else { access.StoreOne(p) } } }, tools.REMOVE_RESOURCE: func(resp tools.NATSResponse) { p := map[string]interface{}{} access := oclib.NewRequestAdmin(oclib.LibDataEnum(resp.Datatype), nil) err := json.Unmarshal(resp.Payload, &p) if err == nil { access.DeleteOne(fmt.Sprintf("%v", p["id"])) } }, }) }