update lightest peer and nats behaviors
This commit is contained in:
@@ -17,7 +17,6 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
oclib "cloud.o-forge.io/core/oc-lib"
|
oclib "cloud.o-forge.io/core/oc-lib"
|
||||||
peer "cloud.o-forge.io/core/oc-lib/models/peer"
|
|
||||||
"github.com/libp2p/go-libp2p/core/host"
|
"github.com/libp2p/go-libp2p/core/host"
|
||||||
"github.com/libp2p/go-libp2p/core/network"
|
"github.com/libp2p/go-libp2p/core/network"
|
||||||
pp "github.com/libp2p/go-libp2p/core/peer"
|
pp "github.com/libp2p/go-libp2p/core/peer"
|
||||||
@@ -72,20 +71,6 @@ func (ix *LongLivedStreamRecordedService[T]) gc() {
|
|||||||
delete(sstreams, pid)
|
delete(sstreams, pid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
ix.PubsubMu.Lock()
|
|
||||||
if ix.LongLivedPubSubs[TopicPubSubNodeActivity] != nil {
|
|
||||||
if b, err := json.Marshal(TopicNodeActivityPub{
|
|
||||||
Disposer: "/ip4/" + conf.GetConfig().Hostname + "/tcp/" + fmt.Sprintf("%v", conf.GetConfig().NodeEndpointPort) + "/p2p/" + ix.Host.ID().String(),
|
|
||||||
Name: rec.HeartbeatStream.Name,
|
|
||||||
DID: rec.HeartbeatStream.DID,
|
|
||||||
PeerID: pid.String(),
|
|
||||||
NodeActivity: peer.OFFLINE.EnumIndex(),
|
|
||||||
}); err == nil {
|
|
||||||
ix.LongLivedPubSubs[TopicPubSubNodeActivity].Publish(context.Background(), b)
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
ix.PubsubMu.Unlock()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -84,7 +84,6 @@ func (pr *PeerRecord) ExtractPeer(ourkey string, key string, pubKey crypto.PubKe
|
|||||||
UUID: pr.DID,
|
UUID: pr.DID,
|
||||||
Name: pr.Name,
|
Name: pr.Name,
|
||||||
},
|
},
|
||||||
State: pp.ONLINE,
|
|
||||||
Relation: rel, // VERIFY.... it crush nothing
|
Relation: rel, // VERIFY.... it crush nothing
|
||||||
PeerID: pr.PeerID,
|
PeerID: pr.PeerID,
|
||||||
PublicKey: base64.StdEncoding.EncodeToString(pubBytes),
|
PublicKey: base64.StdEncoding.EncodeToString(pubBytes),
|
||||||
@@ -93,23 +92,22 @@ func (pr *PeerRecord) ExtractPeer(ourkey string, key string, pubKey crypto.PubKe
|
|||||||
NATSAddress: pr.NATSAddress,
|
NATSAddress: pr.NATSAddress,
|
||||||
WalletAddress: pr.WalletAddress,
|
WalletAddress: pr.WalletAddress,
|
||||||
}
|
}
|
||||||
if time.Now().UTC().After(pr.ExpiryDate) { // is expired
|
|
||||||
p.State = pp.OFFLINE // then is considers OFFLINE
|
|
||||||
}
|
|
||||||
b, err := json.Marshal(p)
|
b, err := json.Marshal(p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return pp.SELF == p.Relation, nil, err
|
return pp.SELF == p.Relation, nil, err
|
||||||
}
|
}
|
||||||
fmt.Println("SENDPEER SELF")
|
|
||||||
go tools.NewNATSCaller().SetNATSPub(tools.CREATE_PEER, tools.NATSResponse{
|
if time.Now().UTC().After(pr.ExpiryDate) {
|
||||||
FromApp: "oc-discovery",
|
|
||||||
Datatype: tools.PEER,
|
|
||||||
Method: int(tools.CREATE_PEER),
|
|
||||||
Payload: b,
|
|
||||||
})
|
|
||||||
if p.State == pp.OFFLINE {
|
|
||||||
return pp.SELF == p.Relation, nil, errors.New("peer " + key + " is offline")
|
return pp.SELF == p.Relation, nil, errors.New("peer " + key + " is offline")
|
||||||
}
|
}
|
||||||
|
go tools.NewNATSCaller().SetNATSPub(tools.CREATE_RESOURCE, tools.NATSResponse{
|
||||||
|
FromApp: "oc-discovery",
|
||||||
|
Datatype: tools.PEER,
|
||||||
|
Method: int(tools.CREATE_RESOURCE),
|
||||||
|
SearchAttr: "peer_id",
|
||||||
|
Payload: b,
|
||||||
|
})
|
||||||
|
|
||||||
return pp.SELF == p.Relation, p, nil
|
return pp.SELF == p.Relation, p, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,19 +5,17 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"oc-discovery/daemons/node/common"
|
"oc-discovery/daemons/node/common"
|
||||||
"oc-discovery/daemons/node/stream"
|
|
||||||
|
|
||||||
oclib "cloud.o-forge.io/core/oc-lib"
|
oclib "cloud.o-forge.io/core/oc-lib"
|
||||||
"cloud.o-forge.io/core/oc-lib/config"
|
"cloud.o-forge.io/core/oc-lib/config"
|
||||||
"cloud.o-forge.io/core/oc-lib/models/peer"
|
"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"
|
"cloud.o-forge.io/core/oc-lib/tools"
|
||||||
pp "github.com/libp2p/go-libp2p/core/peer"
|
pp "github.com/libp2p/go-libp2p/core/peer"
|
||||||
)
|
)
|
||||||
|
|
||||||
func ListenNATS(n *Node) {
|
func ListenNATS(n *Node) {
|
||||||
tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
|
tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
|
||||||
tools.VERIFY_RESOURCE: func(resp tools.NATSResponse) {
|
/*tools.VERIFY_RESOURCE: func(resp tools.NATSResponse) {
|
||||||
if resp.FromApp == config.GetAppName() {
|
if resp.FromApp == config.GetAppName() {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -46,9 +44,9 @@ func ListenNATS(n *Node) {
|
|||||||
n.StreamService.PublishVerifyResources(&resp.Datatype, resp.User, realP.PeerID, resp.Payload)
|
n.StreamService.PublishVerifyResources(&resp.Datatype, resp.User, realP.PeerID, resp.Payload)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},*/
|
||||||
tools.CREATE_PEER: func(resp tools.NATSResponse) {
|
tools.CREATE_RESOURCE: func(resp tools.NATSResponse) {
|
||||||
if resp.FromApp == config.GetAppName() {
|
if resp.FromApp == config.GetAppName() && resp.Datatype != tools.PEER {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
logger := oclib.GetLogger()
|
logger := oclib.GetLogger()
|
||||||
@@ -68,10 +66,6 @@ func ListenNATS(n *Node) {
|
|||||||
n.StreamService.Mu.Lock()
|
n.StreamService.Mu.Lock()
|
||||||
defer n.StreamService.Mu.Unlock()
|
defer n.StreamService.Mu.Unlock()
|
||||||
|
|
||||||
n.Mu.Lock()
|
|
||||||
n.Peers[ad.ID] = p.State == peer.ONLINE
|
|
||||||
n.Mu.Unlock()
|
|
||||||
|
|
||||||
if p.Relation == peer.PARTNER {
|
if p.Relation == peer.PARTNER {
|
||||||
n.StreamService.ConnectToPartner(p.StreamAddress)
|
n.StreamService.ConnectToPartner(p.StreamAddress)
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
@@ -34,7 +34,6 @@ type Node struct {
|
|||||||
isIndexer bool
|
isIndexer bool
|
||||||
|
|
||||||
Mu sync.RWMutex
|
Mu sync.RWMutex
|
||||||
Peers map[pp.ID]bool
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func InitNode(isNode bool, isIndexer bool, isNativeIndexer bool) (*Node, error) {
|
func InitNode(isNode bool, isIndexer bool, isNativeIndexer bool) (*Node, error) {
|
||||||
@@ -65,7 +64,6 @@ func InitNode(isNode bool, isIndexer bool, isNativeIndexer bool) (*Node, error)
|
|||||||
return nil, errors.New("no host no node")
|
return nil, errors.New("no host no node")
|
||||||
}
|
}
|
||||||
node := &Node{
|
node := &Node{
|
||||||
Peers: map[pp.ID]bool{},
|
|
||||||
PeerID: h.ID(),
|
PeerID: h.ID(),
|
||||||
isIndexer: isIndexer,
|
isIndexer: isIndexer,
|
||||||
LongLivedStreamRecordedService: common.NewStreamRecordedService[interface{}](h, 1000, false),
|
LongLivedStreamRecordedService: common.NewStreamRecordedService[interface{}](h, 1000, false),
|
||||||
@@ -101,13 +99,6 @@ func InitNode(isNode bool, isIndexer bool, isNativeIndexer bool) (*Node, error)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
node.SubscribeToSearch(node.PS, &f)
|
node.SubscribeToSearch(node.PS, &f)
|
||||||
access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
|
|
||||||
peers := access.LoadAll(false)
|
|
||||||
for _, p := range peers.Data { // fill cache.
|
|
||||||
if pid, err := pp.Decode(p.(*peer.Peer).PeerID); err == nil {
|
|
||||||
node.Peers[pid] = p.(*peer.Peer).State == peer.ONLINE
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
if isIndexer {
|
if isIndexer {
|
||||||
logger.Info().Msg("generate opencloud indexer...")
|
logger.Info().Msg("generate opencloud indexer...")
|
||||||
@@ -274,3 +265,19 @@ func (d *Node) claimInfo(
|
|||||||
return p, err
|
return p, err
|
||||||
//}
|
//}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
TODO:
|
||||||
|
- Le booking est un flow neuf décentralisé :
|
||||||
|
On check on attend une réponse, on valide, il passe par discovery, on relais.
|
||||||
|
- Le shared workspace est une affaire de décentralisation,
|
||||||
|
on communique avec les shared les mouvements
|
||||||
|
- Un shared remplace la notion de partnership à l'échelle de partnershipping
|
||||||
|
-> quand on share un workspace on devient partenaire temporaire
|
||||||
|
qu'on le soit originellement ou non.
|
||||||
|
-> on a alors les mêmes privilèges.
|
||||||
|
- Les orchestrations admiralty ont le même fonctionnement.
|
||||||
|
Un evenement provoque alors une création de clé de service.
|
||||||
|
|
||||||
|
On doit pouvoir crud avec verification de signature un DBobject.
|
||||||
|
*/
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ type Verify struct {
|
|||||||
|
|
||||||
func (ps *StreamService) handleEvent(protocol string, evt *common.Event) error {
|
func (ps *StreamService) handleEvent(protocol string, evt *common.Event) error {
|
||||||
ps.handleEventFromPartner(evt, protocol)
|
ps.handleEventFromPartner(evt, protocol)
|
||||||
if protocol == ProtocolVerifyResource {
|
/*if protocol == ProtocolVerifyResource {
|
||||||
if evt.DataType == -1 {
|
if evt.DataType == -1 {
|
||||||
tools.NewNATSCaller().SetNATSPub(tools.VERIFY_RESOURCE, tools.NATSResponse{
|
tools.NewNATSCaller().SetNATSPub(tools.VERIFY_RESOURCE, tools.NATSResponse{
|
||||||
FromApp: "oc-discovery",
|
FromApp: "oc-discovery",
|
||||||
@@ -29,7 +29,7 @@ func (ps *StreamService) handleEvent(protocol string, evt *common.Event) error {
|
|||||||
} else if err := ps.verifyResponse(evt); err != nil {
|
} else if err := ps.verifyResponse(evt); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}*/
|
||||||
if protocol == ProtocolSearchResource && evt.DataType > -1 {
|
if protocol == ProtocolSearchResource && evt.DataType > -1 {
|
||||||
if err := ps.retrieveResponse(evt); err != nil {
|
if err := ps.retrieveResponse(evt); err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
@@ -169,7 +169,6 @@ func (s *StreamService) searchPeer(search string) ([]*peer.Peer, error) {
|
|||||||
},
|
},
|
||||||
PeerID: ppID[len(ppID)-1],
|
PeerID: ppID[len(ppID)-1],
|
||||||
StreamAddress: peerID,
|
StreamAddress: peerID,
|
||||||
State: peer.ONLINE,
|
|
||||||
Relation: peer.PARTNER,
|
Relation: peer.PARTNER,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
2
go.mod
2
go.mod
@@ -3,7 +3,7 @@ module oc-discovery
|
|||||||
go 1.24.6
|
go 1.24.6
|
||||||
|
|
||||||
require (
|
require (
|
||||||
cloud.o-forge.io/core/oc-lib v0.0.0-20260212123952-403913d8cf13
|
cloud.o-forge.io/core/oc-lib v0.0.0-20260218132556-0b41e2505e2f
|
||||||
github.com/beego/beego v1.12.13
|
github.com/beego/beego v1.12.13
|
||||||
github.com/beego/beego/v2 v2.3.8
|
github.com/beego/beego/v2 v2.3.8
|
||||||
github.com/go-redis/redis v6.15.9+incompatible
|
github.com/go-redis/redis v6.15.9+incompatible
|
||||||
|
|||||||
2
go.sum
2
go.sum
@@ -48,6 +48,8 @@ cloud.o-forge.io/core/oc-lib v0.0.0-20260209113703-b9c9b6678099 h1:HczicbRtjiU51
|
|||||||
cloud.o-forge.io/core/oc-lib v0.0.0-20260209113703-b9c9b6678099/go.mod h1:jmyBwmsac/4V7XPL347qawF60JsBCDmNAMfn/ySXKYo=
|
cloud.o-forge.io/core/oc-lib v0.0.0-20260209113703-b9c9b6678099/go.mod h1:jmyBwmsac/4V7XPL347qawF60JsBCDmNAMfn/ySXKYo=
|
||||||
cloud.o-forge.io/core/oc-lib v0.0.0-20260212123952-403913d8cf13 h1:DNIPQ7C+7wjbj5RUx29wLxuIe/wiSOcuUMlLRIv6Fvs=
|
cloud.o-forge.io/core/oc-lib v0.0.0-20260212123952-403913d8cf13 h1:DNIPQ7C+7wjbj5RUx29wLxuIe/wiSOcuUMlLRIv6Fvs=
|
||||||
cloud.o-forge.io/core/oc-lib v0.0.0-20260212123952-403913d8cf13/go.mod h1:jmyBwmsac/4V7XPL347qawF60JsBCDmNAMfn/ySXKYo=
|
cloud.o-forge.io/core/oc-lib v0.0.0-20260212123952-403913d8cf13/go.mod h1:jmyBwmsac/4V7XPL347qawF60JsBCDmNAMfn/ySXKYo=
|
||||||
|
cloud.o-forge.io/core/oc-lib v0.0.0-20260218132556-0b41e2505e2f h1:OFuJhi23D/UNwn8Jo30HDt/Sm2Ea1ljUk6IVicYSuAQ=
|
||||||
|
cloud.o-forge.io/core/oc-lib v0.0.0-20260218132556-0b41e2505e2f/go.mod h1:jmyBwmsac/4V7XPL347qawF60JsBCDmNAMfn/ySXKYo=
|
||||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||||
github.com/Knetic/govaluate v3.0.0+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
|
github.com/Knetic/govaluate v3.0.0+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
|
||||||
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
|
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
|
||||||
|
|||||||
Reference in New Issue
Block a user