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

Set up -> Stream is Working

Browse Source
This commit is contained in:
mr
2026-02-03 15:25:15 +01:00
parent 0ff21c0818
commit 1c2ea9ca96
11 changed files with 259 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
daemons/node/stream/handler.go
View File

@@ -40,7 +40,7 @@ func (abs *StreamService) retrieveResponse(event *common.Event) error { //
return nil
}
b, err := json.Marshal(res.Serialize(res))
tools.NewNATSCaller().SetNATSPub(tools.CATALOG_SEARCH_EVENT, tools.NATSResponse{
go tools.NewNATSCaller().SetNATSPub(tools.CATALOG_SEARCH_EVENT, tools.NATSResponse{
FromApp: "oc-discovery",
Datatype: tools.DataType(event.DataType),
Method: int(tools.CATALOG_SEARCH_EVENT),
@@ -74,14 +74,14 @@ func (ps *StreamService) handleEventFromPartner(evt *common.Event, action tools.
}
case tools.PB_CREATE:
case tools.PB_UPDATE:
tools.NewNATSCaller().SetNATSPub(tools.CREATE_RESOURCE, tools.NATSResponse{
go 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{
go tools.NewNATSCaller().SetNATSPub(tools.REMOVE_RESOURCE, tools.NATSResponse{
FromApp: "oc-discovery",
Datatype: tools.DataType(evt.DataType),
Method: int(tools.REMOVE_RESOURCE),

77
daemons/node/stream/service.go
View File

@@ -3,15 +3,18 @@ package stream
import (
"context"
"encoding/json"
"errors"
"fmt"
"oc-discovery/conf"
"oc-discovery/daemons/node/common"
"strings"
"sync"
"time"
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/models/peer"
"cloud.o-forge.io/core/oc-lib/models/utils"
"github.com/google/uuid"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/network"
pp "github.com/libp2p/go-libp2p/core/peer"
@@ -66,11 +69,11 @@ func (s *StreamService) HandlePartnerHeartbeat(stream network.Stream) {
}
s.mu.Lock()
defer s.mu.Unlock()
streams := s.Streams[ProtocolHeartbeatPartner]
if streams == nil {
if s.Streams[ProtocolHeartbeatPartner] == nil {
s.Streams[ProtocolHeartbeatPartner] = map[pp.ID]*common.Stream{}
return
}
streams := s.Streams[ProtocolHeartbeatPartner]
// if record already seen update last seen
if rec, ok := streams[*pid]; ok {
rec.DID = hb.DID
@@ -93,7 +96,7 @@ func (s *StreamService) connectToPartners() error {
for _, p := range peers {
ad, err := pp.AddrInfoFromString(p.StreamAddress)
if err != nil {
return err
continue
}
pid, err := pp.Decode(p.PeerID)
if err != nil {
@@ -102,11 +105,26 @@ func (s *StreamService) connectToPartners() error {
s.ConnectToPartner(pid, ad)
// heartbeat your partner.
}
for _, proto := range protocols {
f := func(ss network.Stream) {
if s.Streams[proto] == nil {
s.Streams[proto] = map[pp.ID]*common.Stream{}
}
s.Streams[proto][ss.Conn().RemotePeer()] = &common.Stream{
Stream: ss,
Expiry: time.Now().UTC().Add(2 * time.Minute),
}
s.readLoop(s.Streams[proto][ss.Conn().RemotePeer()])
}
fmt.Println("SetStreamHandler", proto)
s.Host.SetStreamHandler(proto, f)
}
// TODO if handle... from partner then HeartBeat back
return nil
}
func (s *StreamService) ConnectToPartner(pid pp.ID, ad *pp.AddrInfo) {
logger := oclib.GetLogger()
for _, proto := range protocols {
f := func(ss network.Stream) {
if s.Streams[proto] == nil {
@@ -114,26 +132,43 @@ func (s *StreamService) ConnectToPartner(pid pp.ID, ad *pp.AddrInfo) {
}
s.Streams[proto][pid] = &common.Stream{
Stream: ss,
Expiry: time.Now().Add(2 * time.Minute),
Expiry: time.Now().UTC().Add(2 * time.Minute),
}
go s.readLoop(s.Streams[proto][pid])
s.readLoop(s.Streams[proto][pid])
}
s.Streams = common.AddStreamProtocol(nil, s.Streams, s.Host, proto, pid, s.Key, &f)
if s.Streams[proto][pid] != nil {
go s.readLoop(s.Streams[proto][pid]) // reaaaad...
if s.Host.Network().Connectedness(ad.ID) != network.Connected {
if err := s.Host.Connect(context.Background(), *ad); err != nil {
logger.Err(err)
continue
}
}
s.Streams = common.AddStreamProtocol(nil, s.Streams, s.Host, proto, pid, s.Key, false, &f)
}
common.SendHeartbeat(context.Background(), ProtocolHeartbeatPartner,
s.Host, s.Streams, []*pp.AddrInfo{ad}, time.Minute)
s.Host, s.Streams, []*pp.AddrInfo{ad}, 20*time.Second)
}
func (s *StreamService) searchPeer(search string) ([]*peer.Peer, error) {
/* TODO FOR TEST ONLY A VARS THAT DEFINE ADDRESS... deserialize */
ps := []*peer.Peer{}
if conf.GetConfig().PeerIDS != "" {
for _, peerID := range strings.Split(conf.GetConfig().PeerIDS, ",") {
ppID := strings.Split(peerID, ":")
ps = append(ps, &peer.Peer{
AbstractObject: utils.AbstractObject{
UUID: uuid.New().String(),
Name: ppID[1],
},
PeerID: ppID[1],
StreamAddress: "/ip4/127.0.0.1/tcp/" + ppID[0] + "/p2p/" + ppID[1],
State: peer.ONLINE,
Relation: peer.PARTNER,
})
}
}
access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
peers := access.Search(nil, search, false)
if len(peers.Data) == 0 {
return ps, errors.New("no self available")
}
for _, p := range peers.Data {
ps = append(ps, p.(*peer.Peer))
}
@@ -161,12 +196,12 @@ func (s *StreamService) StartGC(interval time.Duration) {
func (s *StreamService) gc() {
s.mu.Lock()
defer s.mu.Unlock()
now := time.Now()
streams := s.Streams[ProtocolHeartbeatPartner]
if streams == nil {
now := time.Now().UTC()
if s.Streams[ProtocolHeartbeatPartner] == nil {
s.Streams[ProtocolHeartbeatPartner] = map[pp.ID]*common.Stream{}
return
}
streams := s.Streams[ProtocolHeartbeatPartner]
for pid, rec := range streams {
if now.After(rec.Expiry) {
for _, sstreams := range s.Streams {
@@ -180,12 +215,12 @@ func (s *StreamService) gc() {
}
func (ps *StreamService) readLoop(s *common.Stream) {
dec := json.NewDecoder(s.Stream)
defer s.Stream.Close()
for {
var evt common.Event
if err := dec.Decode(&evt); err != nil {
if err := json.NewDecoder(s.Stream).Decode(&evt); err != nil {
s.Stream.Close()
return
continue
}
ps.handleEvent(evt.Type, &evt)
}