core/oc-peer
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fully Working Network Peers

Browse Source
This commit is contained in:
mr
2026-04-01 11:27:05 +02:00
parent 8c6b047ab6
commit bc7b1a2f29
10 changed files with 345 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
infrastructure/nats.go
View File

@@ -18,6 +18,7 @@ var ressourceCols = []oclib.LibDataEnum{
}
var SearchMu sync.RWMutex
var SearchStreamAction = map[string][]*peer.Peer{}
var SearchStream = map[string]chan *peer.Peer{}
func EmitNATS(user string, groups []string, message tools.PropalgationMessage) {
@@ -25,6 +26,7 @@ func EmitNATS(user string, groups []string, message tools.PropalgationMessage) {
if message.Action == tools.PB_SEARCH {
SearchMu.Lock()
SearchStream[user] = make(chan *peer.Peer, 128)
SearchStreamAction[user] = []*peer.Peer{}
SearchMu.Unlock()
}
tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
@@ -55,6 +57,8 @@ func EmitNATS(user string, groups []string, message tools.PropalgationMessage) {
// lui va vérifier la signature de la ressource qu'il possède correspondante si elle existe, si non. AIE,
// on met à jour mais on pète une erreur.
var self *peer.Peer
func ListenNATS() {
tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
tools.CREATE_RESOURCE: func(resp tools.NATSResponse) {
@@ -66,12 +70,19 @@ func ListenNATS() {
/*if err := verify(resp.Payload); err != nil {
return // don't trust anyone... only friends and foes are privilege
}*/
fmt.Println("CREATE_RESOURCE", p.GetID())
if ok, _ := oclib.IsMySelf(p.GetID()); ok {
fmt.Println("it's ourselve !")
return
}
access := oclib.NewRequestAdmin(oclib.LibDataEnum(resp.Datatype), nil)
if data := access.LoadOne(p.GetID()); data.Data != nil {
if p.Relation == peer.PENDING_PARTNER {
if p.Relation == peer.PENDING_PARTNER || p.Relation == peer.PARTNER {
if data.ToPeer().Verify {
fmt.Println("UPDATE 2", p.GetID())
access.UpdateOne(map[string]interface{}{
"verify": false,
"relation": peer.PARTNER,
@@ -82,26 +93,19 @@ func ListenNATS() {
"relation": peer.PENDING_PARTNER,
}, p.GetID())
}
} else if data.ToPeer().Relation != peer.BLACKLIST && data.ToPeer().Relation != peer.SELF {
} else if data.ToPeer().Relation == peer.NONE {
access.UpdateOne(map[string]interface{}{
"verify": false,
"relation": p.Relation,
}, p.GetID())
}
} else if p.Relation != peer.SELF {
if p.Relation != peer.SELF {
access.StoreOne(p.Serialize(p))
} else if data := access.Search(&dbs.Filters{
And: map[string][]dbs.Filter{
"relation": {{Operator: dbs.EQUAL.String(), Value: peer.SELF.EnumIndex()}},
},
}, "", false); len(data.Data) == 0 {
access.StoreOne(p.Serialize(p))
} else if len(data.Data) == 1 && data.Data[0].GetID() != p.GetID() {
access.DeleteOne(data.Data[0].GetID())
access.StoreOne(p.Serialize(p))
} else if p.Relation != peer.SELF && p.Relation != peer.BLACKLIST {
if p.Relation == peer.PARTNER || p.Relation == peer.PENDING_PARTNER {
p.Verify = true
p.Relation = peer.PENDING_PARTNER
}
access.StoreOne(p.Serialize(p))
}
}
},
tools.SEARCH_EVENT: func(resp tools.NATSResponse) {
@@ -111,7 +115,17 @@ func ListenNATS() {
p := &peer.Peer{}
err := json.Unmarshal(resp.Payload, p)
if err == nil {
fmt.Println("ADD in SEARCH STREAM")
access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
fmt.Println("ADD in SEARCH STREAM", p.GetID())
if s := access.Search(&dbs.Filters{
And: map[string][]dbs.Filter{
"peer_id": {{Operator: dbs.EQUAL.String(), Value: p.PeerID}},
},
}, "", false); len(s.Data) > 0 {
p.Relation = s.Data[0].(*peer.Peer).Relation
} else {
p.NotInCatalog = true
}
SearchMu.Lock()
SearchStream[resp.User] <- p // TODO when do we update it in our catalog ?
SearchMu.Unlock()