Change of state Conn Management

Dockerfile

@@ -21,9 +21,6 @@ RUN go mod download
 FROM golang:alpine AS builder
 ARG CONF_NUM
 
-# Fail fast if CONF_NUM missing
-RUN test -n "$CONF_NUM"
-
 RUN apk add --no-cache git
 
 WORKDIR /oc-discovery
@@ -55,13 +52,13 @@ WORKDIR /app
 
 RUN mkdir ./pem
 
-COPY --from=builder /app/extracted/pem/private${CONF_NUM}.pem ./pem/private.pem
+COPY --from=builder /app/extracted/pem/private${CONF_NUM:-1}.pem ./pem/private.pem
 COPY --from=builder /app/extracted/psk ./psk
-COPY --from=builder /app/extracted/pem/public${CONF_NUM}.pem ./pem/public.pem
+COPY --from=builder /app/extracted/pem/public${CONF_NUM:-1}.pem ./pem/public.pem
 
 COPY --from=builder /app/extracted/oc-discovery /usr/bin/oc-discovery
-COPY --from=builder /app/extracted/docker_discovery${CONF_NUM}.json /etc/oc/discovery.json
+COPY --from=builder /app/extracted/docker_discovery${CONF_NUM:-1}.json /etc/oc/discovery.json
 
-EXPOSE 400${CONF_NUM}
+EXPOSE 400${CONF_NUM:-1}
 
 ENTRYPOINT ["oc-discovery"]

Makefile

@@ -10,15 +10,17 @@ clean:
 	rm -rf  oc-discovery
 
 docker:
-	DOCKER_BUILDKIT=1 docker build -t oc/oc-discovery:0.0.1 -f Dockerfile .
+	DOCKER_BUILDKIT=1 docker build -t oc-discovery -f Dockerfile .
-	docker tag oc/oc-discovery:0.0.1 oc/oc-discovery:latest
+	docker tag oc-discovery opencloudregistry/oc-discovery:latest
 
 publish-kind:
 	kind load docker-image oc/oc-discovery:0.0.1 --name opencloud
 
 publish-registry:
-	@echo "TODO"
+	docker push opencloudregistry/oc-discovery:latest
 
-all: docker publish-kind publish-registry
+all: docker publish-kind
+
+ci: docker publish-registry
 
 .PHONY: build run clean docker publish-kind publish-registry

daemons/node/common/common_pubsub.go

@@ -159,20 +159,16 @@ func (s *LongLivedPubSubService) SubscribeToSearch(ps *pubsub.PubSub, f *func(co
 func SubscribeEvents[T interface{}](s *LongLivedPubSubService,
 	ctx context.Context, proto string, timeout int, f func(context.Context, T, string),
 ) error {
-	s.PubsubMu.Lock()
 	if s.LongLivedPubSubs[proto] == nil {
-		s.PubsubMu.Unlock()
 		return errors.New("no protocol subscribed in pubsub")
 	}
 	topic := s.LongLivedPubSubs[proto]
-	s.PubsubMu.Unlock()
-
 	sub, err := topic.Subscribe() // then subscribe to it
 	if err != nil {
 		return err
 	}
 	// launch loop waiting for results.
-	go waitResults[T](s, ctx, sub, proto, timeout, f)
+	go waitResults(s, ctx, sub, proto, timeout, f)
 
 	return nil
 }
@@ -207,10 +203,5 @@ func waitResults[T interface{}](s *LongLivedPubSubService, ctx context.Context,
 			continue
 		}
 		f(ctx, evt, fmt.Sprintf("%v", proto))
-		/*if p, err := ps.Node.GetPeerRecord(ctx, evt.From); err == nil && len(p) > 0 {
-			if err := ps.processEvent(ctx, p[0], &evt, topicName); err != nil {
-				logger.Err(err)
-			}
-		}*/
 	}
 }

daemons/node/common/common_stream.go

@@ -235,7 +235,6 @@ var StreamIndexers ProtocolStream = ProtocolStream{}
 
 func ConnectToIndexers(h host.Host, minIndexer int, maxIndexer int, myPID pp.ID) {
 	logger := oclib.GetLogger()
-	ctx := context.Background()
 	addresses := strings.Split(conf.GetConfig().IndexerAddresses, ",")
 
 	if len(addresses) > maxIndexer {
@@ -243,13 +242,17 @@ func ConnectToIndexers(h host.Host, minIndexer int, maxIndexer int, myPID pp.ID)
 	}
 
 	for _, indexerAddr := range addresses {
+		fmt.Println("GENERATE ADDR", indexerAddr)
 		ad, err := pp.AddrInfoFromString(indexerAddr)
 		if err != nil {
+			fmt.Println("ADDR ERR", err)
 			logger.Err(err)
 			continue
 		}
+		force := false
 		if h.Network().Connectedness(ad.ID) != network.Connected {
-			if err := h.Connect(ctx, *ad); err != nil {
+			force = true
+			if err := h.Connect(context.Background(), *ad); err != nil {
 				fmt.Println(err)
 				logger.Err(err)
 				continue
@@ -258,7 +261,7 @@ func ConnectToIndexers(h host.Host, minIndexer int, maxIndexer int, myPID pp.ID)
 		StaticIndexers = append(StaticIndexers, ad)
 		// make a privilege streams with indexer.
 		for _, proto := range []protocol.ID{ProtocolPublish, ProtocolGet, ProtocolHeartbeat} {
-			AddStreamProtocol(nil, StreamIndexers, h, proto, ad.ID, myPID, true, nil)
+			AddStreamProtocol(nil, StreamIndexers, h, proto, ad.ID, myPID, force, nil)
 		}
 	}
 	if len(StaticIndexers) == 0 {
@@ -268,7 +271,7 @@ func ConnectToIndexers(h host.Host, minIndexer int, maxIndexer int, myPID pp.ID)
 	if len(StaticIndexers) < minIndexer {
 		// TODO : ask for unknown indexer.
 	}
-	SendHeartbeat(ctx, ProtocolHeartbeat, conf.GetConfig().Name, h, StreamIndexers, StaticIndexers, 20*time.Second) // your indexer is just like a node for the next indexer.
+	SendHeartbeat(context.Background(), ProtocolHeartbeat, conf.GetConfig().Name, h, StreamIndexers, StaticIndexers, 20*time.Second) // your indexer is just like a node for the next indexer.
 }
 
 func AddStreamProtocol(ctx *context.Context, protoS ProtocolStream, h host.Host, proto protocol.ID, id pp.ID, mypid pp.ID, force bool, onStreamCreated *func(network.Stream)) ProtocolStream {
@@ -294,7 +297,7 @@ func AddStreamProtocol(ctx *context.Context, protoS ProtocolStream, h host.Host,
 		if protoS[proto][id] != nil {
 			protoS[proto][id].Expiry = time.Now().Add(2 * time.Minute)
 		} else {
-			fmt.Println("GENERATE STREAM", proto, id)
+			fmt.Println("NEW STREAM", proto, id)
 			s, err := h.NewStream(*ctx, id, proto)
 			if err != nil {
 				panic(err.Error())
@@ -385,18 +388,3 @@ func sendHeartbeat(ctx context.Context, h host.Host, proto protocol.ID, p *pp.Ad
 	pss.Expiry = time.Now().UTC().Add(2 * time.Minute)
 	return nil
 }
-
-/*
-func SearchPeer(search string) ([]*peer.Peer, error) {
-	ps := []*peer.Peer{}
-	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))
-	}
-	return ps, nil
-}
-*/

daemons/node/indexer/handler.go

@@ -79,7 +79,6 @@ func (pr *PeerRecord) ExtractPeer(ourkey string, key string, pubKey crypto.PubKe
 	if err != nil {
 		return false, nil, err
 	}
-	fmt.Println("ExtractPeer MarshalPublicKey")
 	rel := pp.NONE
 	if ourkey == key { // at this point is PeerID is same as our... we are... thats our peer INFO
 		rel = pp.SELF
@@ -106,7 +105,8 @@ func (pr *PeerRecord) ExtractPeer(ourkey string, key string, pubKey crypto.PubKe
 	if err != nil {
 		return pp.SELF == p.Relation, nil, err
 	}
-	go tools.NewNATSCaller().SetNATSPub(tools.CREATE_RESOURCE, tools.NATSResponse{
+	fmt.Println("SENDPEER SELF")
+	go tools.NewNATSCaller().SetNATSPub(tools.CREATE_PEER, tools.NATSResponse{
 		FromApp:  "oc-discovery",
 		Datatype: tools.PEER,
 		Method:   int(tools.CREATE_PEER),
@@ -128,6 +128,7 @@ type GetResponse struct {
 }
 
 func (ix *IndexerService) initNodeHandler() {
+	fmt.Println("Node activity")
 	ix.Host.SetStreamHandler(common.ProtocolHeartbeat, ix.HandleNodeHeartbeat)
 	ix.Host.SetStreamHandler(common.ProtocolPublish, ix.handleNodePublish)
 	ix.Host.SetStreamHandler(common.ProtocolGet, ix.handleNodeGet)

daemons/node/nats.go

@@ -4,12 +4,53 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"oc-discovery/daemons/node/common"
 
+	oclib "cloud.o-forge.io/core/oc-lib"
+	"cloud.o-forge.io/core/oc-lib/config"
+	"cloud.o-forge.io/core/oc-lib/models/peer"
 	"cloud.o-forge.io/core/oc-lib/tools"
+	pp "github.com/libp2p/go-libp2p/core/peer"
 )
 
 func ListenNATS(n Node) {
 	tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
+		tools.CREATE_PEER: func(resp tools.NATSResponse) {
+			if resp.FromApp == config.GetAppName() {
+				return
+			}
+			logger := oclib.GetLogger()
+			m := map[string]interface{}{}
+			err := json.Unmarshal(resp.Payload, &m)
+			if err != nil {
+				logger.Err(err)
+				return
+			}
+			p := &peer.Peer{}
+			p = p.Deserialize(m, p).(*peer.Peer)
+
+			ad, err := pp.AddrInfoFromString(p.PeerID)
+			n.StreamService.Mu.Lock()
+			defer n.StreamService.Mu.Unlock()
+
+			if p.Relation == peer.PARTNER {
+				n.StreamService.ConnectToPartner(ad.ID, ad)
+			} else {
+				ps := common.ProtocolStream{}
+				for p, s := range n.StreamService.Streams {
+					m := map[pp.ID]*common.Stream{}
+					for k := range s {
+						if ad.ID != k {
+							m[k] = s[k]
+						} else {
+							s[k].Stream.Close()
+						}
+					}
+					ps[p] = m
+				}
+				n.StreamService.Streams = ps
+			}
+		},
 		tools.PROPALGATION_EVENT: func(resp tools.NATSResponse) {
 			var propalgation tools.PropalgationMessage
 			err := json.Unmarshal(resp.Payload, &propalgation)

daemons/node/stream/publish.go

@@ -91,8 +91,8 @@ func (ps *StreamService) ToPartnerPublishEvent(
 		if err != nil {
 			return err
 		}
-		ps.mu.Lock()
+		ps.Mu.Lock()
-		defer ps.mu.Unlock()
+		defer ps.Mu.Unlock()
 		if p.Relation == peer.PARTNER {
 			if ps.Streams[ProtocolHeartbeatPartner] == nil {
 				ps.Streams[ProtocolHeartbeatPartner] = map[pp.ID]*common.Stream{}
@@ -138,8 +138,8 @@ func (s *StreamService) write(
 	if dt != nil {
 		name = action.String() + "." + (*dt).String() + "#" + peerID.ID.String()
 	}
-	s.mu.Lock()
+	s.Mu.Lock()
-	defer s.mu.Unlock()
+	defer s.Mu.Unlock()
 	if s.Streams[proto] == nil {
 		s.Streams[proto] = map[pp.ID]*common.Stream{}
 	}

daemons/node/stream/service.go

@@ -41,7 +41,7 @@ type StreamService struct {
 	Node         common.DiscoveryPeer
 	Streams      common.ProtocolStream
 	maxNodesConn int
-	mu           sync.Mutex
+	Mu           sync.Mutex
 	// Stream       map[protocol.ID]map[pp.ID]*daemons.Stream
 }
 
@@ -67,8 +67,8 @@ func (s *StreamService) HandlePartnerHeartbeat(stream network.Stream) {
 	if err != nil {
 		return
 	}
-	s.mu.Lock()
+	s.Mu.Lock()
-	defer s.mu.Unlock()
+	defer s.Mu.Unlock()
 
 	if s.Streams[ProtocolHeartbeatPartner] == nil {
 		s.Streams[ProtocolHeartbeatPartner] = map[pp.ID]*common.Stream{}
@@ -89,6 +89,20 @@ func (s *StreamService) HandlePartnerHeartbeat(stream network.Stream) {
 }
 
 func (s *StreamService) connectToPartners() error {
+	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),
+			}
+			go s.readLoop(s.Streams[proto][ss.Conn().RemotePeer()])
+		}
+		fmt.Println("SetStreamHandler", proto)
+		s.Host.SetStreamHandler(proto, f)
+	}
 	peers, err := s.searchPeer(fmt.Sprintf("%v", peer.PARTNER.EnumIndex()))
 	if err != nil {
 		return err
@@ -105,26 +119,13 @@ 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()
+	force := false
 	for _, proto := range protocols {
 		f := func(ss network.Stream) {
 			if s.Streams[proto] == nil {
@@ -134,15 +135,16 @@ func (s *StreamService) ConnectToPartner(pid pp.ID, ad *pp.AddrInfo) {
 				Stream: ss,
 				Expiry: time.Now().UTC().Add(2 * time.Minute),
 			}
-			s.readLoop(s.Streams[proto][pid])
+			go s.readLoop(s.Streams[proto][pid])
 		}
 		if s.Host.Network().Connectedness(ad.ID) != network.Connected {
+			force = true
 			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)
+		s.Streams = common.AddStreamProtocol(nil, s.Streams, s.Host, proto, pid, s.Key, force, &f)
 	}
 	common.SendHeartbeat(context.Background(), ProtocolHeartbeatPartner, conf.GetConfig().Name,
 		s.Host, s.Streams, []*pp.AddrInfo{ad}, 20*time.Second)
@@ -153,14 +155,15 @@ func (s *StreamService) searchPeer(search string) ([]*peer.Peer, error) {
 	ps := []*peer.Peer{}
 	if conf.GetConfig().PeerIDS != "" {
 		for _, peerID := range strings.Split(conf.GetConfig().PeerIDS, ",") {
-			ppID := strings.Split(peerID, ":")
+			ppID := strings.Split(peerID, "/")
+			fmt.Println(ppID, peerID)
 			ps = append(ps, &peer.Peer{
 				AbstractObject: utils.AbstractObject{
 					UUID: uuid.New().String(),
 					Name: ppID[1],
 				},
-				PeerID:        ppID[1],
+				PeerID:        ppID[len(ppID)-1],
-				StreamAddress: "/ip4/127.0.0.1/tcp/" + ppID[0] + "/p2p/" + ppID[1],
+				StreamAddress: peerID,
 				State:         peer.ONLINE,
 				Relation:      peer.PARTNER,
 			})
@@ -194,8 +197,8 @@ func (s *StreamService) StartGC(interval time.Duration) {
 }
 
 func (s *StreamService) gc() {
-	s.mu.Lock()
+	s.Mu.Lock()
-	defer s.mu.Unlock()
+	defer s.Mu.Unlock()
 	now := time.Now().UTC()
 
 	if s.Streams[ProtocolHeartbeatPartner] == nil {

docker_discovery2.json

@@ -4,5 +4,5 @@
     "NATS_URL": "nats://nats:4222",
     "NODE_MODE": "indexer",
     "NODE_ENDPOINT_PORT": 4002,
-    "INDEXER_ADDRESSES": "/ip4/oc-discovery1/tcp/4001/p2p/12D3KooWGn3j4XqTSrjJDGGpTQERdDV5TPZdhQp87rAUnvQssvQu"
+    "INDEXER_ADDRESSES": "/ip4/172.19.0.2/tcp/4001/p2p/12D3KooWGn3j4XqTSrjJDGGpTQERdDV5TPZdhQp87rAUnvQssvQu"
 }

docker_discovery3.json

@@ -4,5 +4,5 @@
     "NATS_URL": "nats://nats:4222",
     "NODE_MODE": "node",
     "NODE_ENDPOINT_PORT": 4003,
-    "INDEXER_ADDRESSES": "/ip4/oc-discovery2/tcp/4002/p2p/12D3KooWC3GNStak8KCYtJq11Dxiq45EJV53z1ZvKetMcZBeBX6u"
+    "INDEXER_ADDRESSES": "/ip4/172.19.0.3/tcp/4002/p2p/12D3KooWC3GNStak8KCYtJq11Dxiq45EJV53z1ZvKetMcZBeBX6u"
 }

docker_discovery4.json

@@ -4,6 +4,6 @@
     "NATS_URL": "nats://nats:4222",
     "NODE_MODE": "node",
     "NODE_ENDPOINT_PORT": 4004,
-    "INDEXER_ADDRESSES": "/ip4/oc-discovery1/tcp/4001/p2p/12D3KooWGn3j4XqTSrjJDGGpTQERdDV5TPZdhQp87rAUnvQssvQu",
+    "INDEXER_ADDRESSES": "/ip4/172.19.0.2/tcp/4001/p2p/12D3KooWGn3j4XqTSrjJDGGpTQERdDV5TPZdhQp87rAUnvQssvQu",
-    "PEER_IDS": "/ip4/oc-discovery3/tcp/4003/p2p/12D3KooWGn3j4XqTSrjJDGGpTQERdDV5TPZdhQp87rAUnvQssvQu"
+    "PEER_IDS": "/ip4/172.19.0.4/tcp/4003/p2p/12D3KooWBh9kZrekBAE5G33q4jCLNRAzygem3gP1mMdK8mhoCTaw"
 }

go.mod

@@ -3,7 +3,7 @@ module oc-discovery
 go 1.24.6
 
 require (
-	cloud.o-forge.io/core/oc-lib v0.0.0-20260203150531-ef916fe2d995
+	cloud.o-forge.io/core/oc-lib v0.0.0-20260205131630-342451db2581
 	github.com/beego/beego v1.12.13
 	github.com/beego/beego/v2 v2.3.8
 	github.com/go-redis/redis v6.15.9+incompatible

go.sum

@@ -34,6 +34,10 @@ cloud.o-forge.io/core/oc-lib v0.0.0-20260203150123-4258f6b58083 h1:nKiU4AfeX+axS
 cloud.o-forge.io/core/oc-lib v0.0.0-20260203150123-4258f6b58083/go.mod h1:T0UCxRd8w+qCVVC0NEyDiWIGC5ADwEbQ7hFcvftd4Ks=
 cloud.o-forge.io/core/oc-lib v0.0.0-20260203150531-ef916fe2d995 h1:ZDRvnzTTNHgMm5hYmseHdEPqQ6rn/4v+P9f/JIxPaNw=
 cloud.o-forge.io/core/oc-lib v0.0.0-20260203150531-ef916fe2d995/go.mod h1:T0UCxRd8w+qCVVC0NEyDiWIGC5ADwEbQ7hFcvftd4Ks=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260205131048-425cd2a9ba2f h1:Ku6u+SeoNXHMBzckekGyXCHLDJPh20Y8GayO6fXEcZE=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260205131048-425cd2a9ba2f/go.mod h1:T0UCxRd8w+qCVVC0NEyDiWIGC5ADwEbQ7hFcvftd4Ks=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260205131630-342451db2581 h1:V9eANWFEkoEPg3nWCvYXnLYbKDdAm3/Y7uCw1nt22Cc=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260205131630-342451db2581/go.mod h1:T0UCxRd8w+qCVVC0NEyDiWIGC5ADwEbQ7hFcvftd4Ks=
 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/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=

main.go

@@ -21,7 +21,7 @@ func main() {
 	oclib.InitDaemon(appname)
 	// get the right config file
 
-	o := oclib.GetConfLoader()
+	o := oclib.GetConfLoader(appname)
 
 	conf.GetConfig().Name = o.GetStringDefault("NAME", "opencloud-demo")
 	conf.GetConfig().Hostname = o.GetStringDefault("HOSTNAME", "127.0.0.1")