124 lines
3.7 KiB
Go
124 lines
3.7 KiB
Go
package stream
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"oc-discovery/daemons/node/common"
|
|
"strings"
|
|
|
|
oclib "cloud.o-forge.io/core/oc-lib"
|
|
"cloud.o-forge.io/core/oc-lib/models/peer"
|
|
"cloud.o-forge.io/core/oc-lib/models/resources"
|
|
"cloud.o-forge.io/core/oc-lib/tools"
|
|
)
|
|
|
|
func (ps *StreamService) getTopicName(topicName string) tools.PubSubAction {
|
|
ns := strings.Split(topicName, ".")
|
|
if len(ns) > 0 {
|
|
return tools.GetActionString(ns[0])
|
|
}
|
|
return tools.NONE
|
|
}
|
|
|
|
func (ps *StreamService) handleEvent(topicName string, evt *common.Event) error {
|
|
action := ps.getTopicName(topicName)
|
|
if err := ps.handleEventFromPartner(evt, action); err != nil {
|
|
return err
|
|
}
|
|
if action == tools.PB_SEARCH_RESPONSE {
|
|
if err := ps.retrieveResponse(evt); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (abs *StreamService) retrieveResponse(event *common.Event) error { //
|
|
res, err := resources.ToResource(int(event.DataType), event.Payload)
|
|
if err != nil || res == nil {
|
|
return nil
|
|
}
|
|
b, err := json.Marshal(res.Serialize(res))
|
|
tools.NewNATSCaller().SetNATSPub(tools.CATALOG_SEARCH_EVENT, tools.NATSResponse{
|
|
FromApp: "oc-discovery",
|
|
Datatype: tools.DataType(event.DataType),
|
|
Method: int(tools.CATALOG_SEARCH_EVENT),
|
|
Payload: b,
|
|
})
|
|
return nil
|
|
}
|
|
|
|
func (ps *StreamService) handleEventFromPartner(evt *common.Event, action tools.PubSubAction) error {
|
|
if !(action == tools.PB_CREATE || action == tools.PB_UPDATE || action == tools.PB_DELETE) {
|
|
return nil
|
|
}
|
|
resource, err := resources.ToResource(int(evt.DataType), evt.Payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
b, err := json.Marshal(resource)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch action {
|
|
case tools.PB_SEARCH:
|
|
access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
|
|
peers := access.Search(nil, evt.From, false)
|
|
if len(peers.Data) > 0 {
|
|
p := peers.Data[0].(*peer.Peer)
|
|
// TODO : something if peer is missing in our side !
|
|
ps.SendResponse(p, evt)
|
|
} else if p, err := ps.Node.GetPeerRecord(context.Background(), evt.From); err == nil {
|
|
ps.SendResponse(p, evt)
|
|
}
|
|
case tools.PB_CREATE:
|
|
case tools.PB_UPDATE:
|
|
tools.NewNATSCaller().SetNATSPub(tools.CREATE_RESOURCE, tools.NATSResponse{
|
|
FromApp: "oc-discovery",
|
|
Datatype: tools.DataType(evt.DataType),
|
|
Method: int(tools.CREATE_RESOURCE),
|
|
Payload: b,
|
|
})
|
|
case tools.PB_DELETE:
|
|
tools.NewNATSCaller().SetNATSPub(tools.REMOVE_RESOURCE, tools.NATSResponse{
|
|
FromApp: "oc-discovery",
|
|
Datatype: tools.DataType(evt.DataType),
|
|
Method: int(tools.REMOVE_RESOURCE),
|
|
Payload: b,
|
|
})
|
|
default:
|
|
return errors.New("no action authorized available : " + action.String())
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (abs *StreamService) SendResponse(p *peer.Peer, event *common.Event) error {
|
|
dts := []oclib.LibDataEnum{oclib.LibDataEnum(event.DataType)}
|
|
if event.DataType == -1 { // expect all resources
|
|
dts = []oclib.LibDataEnum{oclib.LibDataEnum(oclib.COMPUTE_RESOURCE), oclib.LibDataEnum(oclib.STORAGE_RESOURCE),
|
|
oclib.LibDataEnum(oclib.PROCESSING_RESOURCE), oclib.LibDataEnum(oclib.DATA_RESOURCE), oclib.LibDataEnum(oclib.WORKFLOW_RESOURCE)}
|
|
}
|
|
var m map[string]string
|
|
err := json.Unmarshal(event.Payload, &m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, dt := range dts {
|
|
access := oclib.NewRequestAdmin(oclib.LibDataEnum(event.DataType), nil)
|
|
peerID := p.GetID()
|
|
searched := access.Search(abs.FilterPeer(peerID, m["search"]), "", false)
|
|
for _, ss := range searched.Data {
|
|
if j, err := json.Marshal(ss); err == nil {
|
|
if event.DataType != -1 {
|
|
ndt := tools.DataType(dt.EnumIndex())
|
|
abs.PublishResources(&ndt, event.User, peerID, j)
|
|
} else {
|
|
abs.PublishResources(nil, event.User, peerID, j)
|
|
} // TODO : TEMP STREAM !
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|