Merge pull request #297 from meshplus/fix-swarm-connect

fix:fix swarm connect with less peers

pkg/order/syncer/syncer_test.go

@@ -2,7 +2,6 @@ package syncer
 
 import (
 	"fmt"
-	"github.com/meshplus/bitxhub-kit/types"
 	"io/ioutil"
 	"strings"
 	"testing"
@@ -15,6 +14,7 @@ import (
 	"github.com/meshplus/bitxhub-kit/crypto/asym"
 	"github.com/meshplus/bitxhub-kit/crypto/asym/ecdsa"
 	"github.com/meshplus/bitxhub-kit/log"
+	"github.com/meshplus/bitxhub-kit/types"
 	"github.com/meshplus/bitxhub-model/pb"
 	"github.com/meshplus/bitxhub/internal/ledger/mock_ledger"
 	"github.com/meshplus/bitxhub/internal/repo"
@@ -27,6 +27,9 @@ func TestStateSyncer_SyncCFTBlocks(t *testing.T) {
 	peerCnt := 3
 	swarms := NewSwarms(t, peerCnt)
 
+	for swarms[0].CountConnectedPeers() != 2 {
+		time.Sleep(100*time.Millisecond)
+	}
 	otherPeers := swarms[0].OtherPeers()
 	peerIds := make([]uint64, 0)
 	for id, _ := range otherPeers {
@@ -57,7 +60,9 @@ func TestStateSyncer_SyncBFTBlocks(t *testing.T) {
 	peerCnt := 4
 	swarms := NewSwarms(t, peerCnt)
 
-	//time.Sleep(100 * time.Millisecond)
+	for swarms[0].CountConnectedPeers() != 3 {
+		time.Sleep(100*time.Millisecond)
+	}
 	otherPeers := swarms[0].OtherPeers()
 	peerIds := make([]uint64, 0)
 	for id, _ := range otherPeers {
@@ -180,10 +185,6 @@ func NewSwarms(t *testing.T, peerCnt int) []*peermgr.Swarm {
 				CACert:         cert,
 			},
 			Config: &repo.Config{
-				Ping: repo.Ping{
-					Enable:   true,
-					Duration: 2 * time.Second,
-				},
 			},
 		}
 

pkg/peermgr/handle_test.go

@@ -163,6 +163,9 @@ func TestSwarm_Send(t *testing.T) {
 	peerCnt := 4
 	swarms := NewSwarms(t, peerCnt)
 
+	for swarms[0].CountConnectedPeers() != 3 {
+		time.Sleep(100*time.Millisecond)
+	}
 
 	msg := &pb.Message{
 		Type: pb.Message_GET_BLOCK,
@@ -297,6 +300,10 @@ func TestSwarm_AsyncSend(t *testing.T) {
 	peerCnt := 4
 	swarms := NewSwarms(t, peerCnt)
 
+	for swarms[0].CountConnectedPeers() != 3 {
+		time.Sleep(100*time.Millisecond)
+	}
+
 	orderMsgCh := make(chan events.OrderMessageEvent)
 	orderMsgSub := swarms[2].SubscribeOrderMessage(orderMsgCh)
 

pkg/peermgr/peermgr.go

@@ -28,6 +28,9 @@ type PeerManager interface {
 	// Broadcast message to all node
 	Broadcast(*pb.Message) error
 
+	// CountConnectedPeers counts connected peer numbers
+	CountConnectedPeers() uint64
+
 	// Peers
 	Peers() map[uint64]*pb.VpInfo
 
@@ -44,7 +47,7 @@ type PeerManager interface {
 	DelNode(delID uint64)
 
 	// UpdateRouter update the local router to quorum router.
-	UpdateRouter (vpInfos map[uint64]*pb.VpInfo, isNew bool) bool
+	UpdateRouter(vpInfos map[uint64]*pb.VpInfo, isNew bool) bool
 
 	// Disconnect disconnect with all vp peers.
 	Disconnect(vpInfos map[uint64]*pb.VpInfo)

pkg/peermgr/swarm.go

@@ -18,8 +18,8 @@ import (
 	"github.com/meshplus/bitxhub/internal/repo"
 	"github.com/meshplus/bitxhub/pkg/cert"
 	network "github.com/meshplus/go-lightp2p"
-	"github.com/sirupsen/logrus"
 	ma "github.com/multiformats/go-multiaddr"
+	"github.com/sirupsen/logrus"
 )
 
 const (
@@ -467,6 +467,15 @@ func (swarm *Swarm) Disconnect(vpInfos map[uint64]*pb.VpInfo) {
 	}
 }
 
+func (swarm *Swarm) CountConnectedPeers() uint64 {
+	var counter uint64
+	swarm.connectedPeers.Range(func(k, v interface{}) bool {
+		counter++
+		return true
+	})
+	return counter
+}
+
 func (swarm *Swarm) reset() {
 	swarm.routers = nil
 	swarm.multiAddrs = nil
@@ -475,7 +484,7 @@ func (swarm *Swarm) reset() {
 }
 
 func constructMultiaddr(vpInfo *pb.VpInfo) (*peer.AddrInfo, error) {
-	addrs := make([]ma.Multiaddr,0)
+	addrs := make([]ma.Multiaddr, 0)
 	for _, host := range vpInfo.Hosts {
 		addr, err := ma.NewMultiaddr(fmt.Sprintf("%s%s", host, vpInfo.Pid))
 		if err != nil {
@@ -490,4 +499,3 @@ func constructMultiaddr(vpInfo *pb.VpInfo) (*peer.AddrInfo, error) {
 	}
 	return addrInfo, nil
 }
-