Fixed bug

internal/peermgr/swarm.go

@@ -3,9 +3,6 @@ package peermgr
 import (
 	"context"
 	"fmt"
-	"github.com/link33/sidecar/internal/port"
-	"github.com/link33/sidecar/internal/router"
-	"github.com/link33/sidecar/model/pb"
 	"strings"
 	"sync"
 	"time"
@@ -15,7 +12,10 @@ import (
 	"github.com/ipfs/go-cid"
 	crypto2 "github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/link33/sidecar/internal/port"
 	"github.com/link33/sidecar/internal/repo"
+	"github.com/link33/sidecar/internal/router"
+	"github.com/link33/sidecar/model/pb"
 	"github.com/meshplus/bitxhub-kit/crypto"
 	"github.com/meshplus/bitxhub-kit/crypto/asym/ecdsa"
 	network "github.com/meshplus/go-lightp2p"
@@ -91,6 +91,13 @@ func New(config *repo.Config, router router.Router, nodePrivKey crypto.PrivateKe
 	}, nil
 }
 
+func (swarm *Swarm) add(p port.Port) {
+	err := swarm.router.Add(p)
+	if err != nil {
+		swarm.logger.Error(err)
+	}
+}
+
 func (swarm *Swarm) Start() error {
 	l := &local{
 		id:      swarm.id,
@@ -99,18 +106,14 @@ func (swarm *Swarm) Start() error {
 		rev:     make(chan *pb.Message),
 		rout:    swarm.router,
 	}
-	swarm.router.Add(l)
-
+	swarm.add(l)
 	swarm.p2p.SetMessageHandler(swarm.handleMessage)
-
 	if err := swarm.RegisterMsgHandler(pb.Message_ADDRESS_GET, swarm.handleGetAddressMessage); err != nil {
 		return fmt.Errorf("register get address msg handler: %w", err)
 	}
-
 	if err := swarm.RegisterMsgHandler(pb.Message_PEER_INFO_GET, l.HandleGetPeerInfoMessage); err != nil {
 		return fmt.Errorf("register get peer info msg handler: %w", err)
 	}
-
 	if err := swarm.RegisterMultiMsgHandler([]pb.Message_Type{
 		pb.Message_IBTP_SEND,
 		pb.Message_IBTP_GET,
@@ -155,14 +158,7 @@ func (swarm *Swarm) Start() error {
 					"address:": address,
 				}).Info("Connect successfully")
 
-				rec := make(chan *pb.Message)
-				p := &sidecar{
-					id:    id,
-					swarm: swarm,
-					tag:   "",
-					rev:   rec,
-				}
-				_ = swarm.router.Add(p)
+				swarm.addRemotePortByID(id)
 
 				swarm.lock.RLock()
 				defer swarm.lock.RUnlock()
@@ -219,6 +215,7 @@ func (swarm *Swarm) handleMessage(s network.Stream, data []byte) {
 		tag:   "",
 		rev:   rec,
 	}
+
 	msgHandler(p, m)
 }
 
@@ -232,14 +229,18 @@ func (swarm *Swarm) HandleIBTPX(pt port.Port, m *pb.Message) {
 		}
 	} else {
 		ppt := pt.(*sidecar)
+		swarm.addRemotePort(ppt)
+		ppt.rev <- m
+	}
+}
+
+func (swarm *Swarm) addRemotePort(ppt *sidecar) {
 	info, err := swarm.getRemotePeerInfo(ppt.ID())
 	if err != nil {
 		swarm.logger.Error(err)
 	}
 	ppt.id = info.GetID()
-		_ = swarm.router.Add(ppt)
-		ppt.rev <- m
-	}
+	swarm.add(ppt)
 }
 
 func (swarm *Swarm) Stop() error {
@@ -264,15 +265,20 @@ func (swarm *Swarm) Connect(addrInfo *peer.AddrInfo) (string, error) {
 		"addrInfo": addrInfo,
 	}).Info("Connect peer")
 
+	id := addrInfo.ID.String()
+	swarm.addRemotePortByID(id)
+	return id, nil
+}
+
+func (swarm *Swarm) addRemotePortByID(id string) {
 	rec := make(chan *pb.Message)
 	p := &sidecar{
-		id:    addrInfo.ID.String(),
+		id:    id,
 		swarm: swarm,
 		tag:   "",
 		rev:   rec,
 	}
-	swarm.router.Add(p)
-	return addrInfo.ID.String(), nil
+	swarm.addRemotePort(p)
 }
 
 func (swarm *Swarm) AsyncSendWithPort(s port.Port, msg *pb.Message) error {
@@ -401,7 +407,7 @@ func (swarm *Swarm) getRemotePeerInfo(id string) (*pb.PeerInfo, error) {
 	if err != nil {
 		return nil, err
 	}
-	retData, err := swarm.p2p.Send(id.String(), reqData) //同步获取数据
+	retData, err := swarm.p2p.Send(id, reqData) //同步获取数据
 	if err != nil {
 		return nil, fmt.Errorf("sync send: %w", err)
 	}