Merge pull request #119 from harmony-one/HAR-67_libp2p_integration_v2

[HAR-67]: libp2p integration v2

benchmark.go

@@ -225,5 +225,5 @@ func main() {
 	}
 
 	go currentNode.SupportSyncing()
-	currentNode.StartServer(*port)
+	currentNode.StartServer()
 }

client/txgen/main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"flag"
 	"fmt"
-	"github.com/harmony-one/harmony/utils"
 	"os"
 	"path"
 	"runtime"
@@ -21,6 +20,7 @@ import (
 	"github.com/harmony-one/harmony/node"
 	"github.com/harmony-one/harmony/p2p"
 	proto_node "github.com/harmony-one/harmony/proto/node"
+	"github.com/harmony-one/harmony/utils"
 )
 
 var (
@@ -89,56 +89,55 @@ func main() {
 	}
 
 	// Client/txgenerator server node setup
+	if clientPeer == nil {
+		panic("Client Peer is nil!")
+	}
 	consensusObj := consensus.New(*clientPeer, "0", nil, p2p.Peer{})
 	clientNode := node.New(consensusObj, nil, *clientPeer)
 
-	if clientPeer != nil {
+	clientNode.Client = client.NewClient(&shardIDLeaderMap)
-		//p2pv2.InitHost(clientPeer.IP, clientPeer.Port) // TODO: this should be moved into client node.
+
-		clientNode.Client = client.NewClient(&shardIDLeaderMap)
+	// This func is used to update the client's utxopool when new blocks are received from the leaders
-
+	updateBlocksFunc := func(blocks []*blockchain.Block) {
-		// This func is used to update the client's utxopool when new blocks are received from the leaders
+		log.Debug("Received new block from leader", "len", len(blocks))
-		updateBlocksFunc := func(blocks []*blockchain.Block) {
+		for _, block := range blocks {
-			log.Debug("Received new block from leader", "len", len(blocks))
+			for _, node := range nodes {
-			for _, block := range blocks {
+				shardID := block.ShardID
-				for _, node := range nodes {
-					shardID := block.ShardID
-
-					accountBlock := new(types.Block)
-					err := rlp.DecodeBytes(block.AccountBlock, accountBlock)
-					if err == nil {
-						shardID = accountBlock.ShardID()
-					}
-					if node.Consensus.ShardID == shardID {
-						log.Debug("Adding block from leader", "shardID", shardID)
-						// Add it to blockchain
-						node.AddNewBlock(block)
-						utxoPoolMutex.Lock()
-						node.UpdateUtxoAndState(block)
-						utxoPoolMutex.Unlock()
 
+				accountBlock := new(types.Block)
+				err := rlp.DecodeBytes(block.AccountBlock, accountBlock)
+				if err == nil {
+					shardID = accountBlock.ShardID()
+				}
+				if node.Consensus.ShardID == shardID {
+					log.Debug("Adding block from leader", "shardID", shardID)
+					// Add it to blockchain
+					node.AddNewBlock(block)
+					utxoPoolMutex.Lock()
+					node.UpdateUtxoAndState(block)
+					utxoPoolMutex.Unlock()
+
+					if err != nil {
+						log.Error("Failed decoding the block with RLP")
+					} else {
+						fmt.Println("RECEIVED NEW BLOCK ", len(accountBlock.Transactions()))
+						node.AddNewBlockAccount(accountBlock)
+						node.Worker.UpdateCurrent()
 						if err != nil {
-							log.Error("Failed decoding the block with RLP")
+							log.Debug("Failed to add new block to worker", "Error", err)
-						} else {
-							fmt.Println("RECEIVED NEW BLOCK ", len(accountBlock.Transactions()))
-							node.AddNewBlockAccount(accountBlock)
-							node.Worker.UpdateCurrent()
-							if err != nil {
-								log.Debug("Failed to add new block to worker", "Error", err)
-							}
 						}
-					} else {
-						continue
 					}
+				} else {
+					continue
 				}
 			}
 		}
-		clientNode.Client.UpdateBlocks = updateBlocksFunc
-
-		// Start the client server to listen to leader's message
-		go func() {
-			clientNode.StartServer(clientPeer.Port)
-		}()
 	}
+	clientNode.Client.UpdateBlocks = updateBlocksFunc
+
+	// Start the client server to listen to leader's message
+	go clientNode.StartServer()
+
 	// Transaction generation process
 	time.Sleep(10 * time.Second) // wait for nodes to be ready
 	start := time.Now()

client/wallet/main.go

@@ -93,7 +93,7 @@ func main() {
 			fmt.Println("Private key imported...")
 		case "showBalance":
 			walletNode := CreateWalletServerNode()
-			go walletNode.StartServer(walletNode.ClientPeer.Port)
+			go walletNode.StartServer()
 
 			shardUtxoMap, err := FetchUtxos(ReadAddresses(), walletNode)
 			if err != nil {
@@ -163,7 +163,7 @@ func main() {
 
 		// Start client server
 		walletNode := CreateWalletServerNode()
-		go walletNode.StartServer(walletNode.ClientPeer.Port)
+		go walletNode.StartServer()
 
 		shardUtxoMap, err := FetchUtxos([][20]byte{senderAddressBytes}, walletNode)
 		if err != nil {

client/wallet_v2/main.go

@@ -93,7 +93,7 @@ func main() {
 			fmt.Println("Private key imported...")
 		case "showBalance":
 			walletNode := CreateWalletServerNode()
-			go walletNode.StartServer(walletNode.ClientPeer.Port)
+			go walletNode.StartServer()
 
 			shardUtxoMap, err := FetchUtxos(ReadAddresses(), walletNode)
 			if err != nil {
@@ -163,7 +163,7 @@ func main() {
 
 		// Start client server
 		walletNode := CreateWalletServerNode()
-		go walletNode.StartServer(walletNode.ClientPeer.Port)
+		go walletNode.StartServer()
 
 		shardUtxoMap, err := FetchUtxos([][20]byte{senderAddressBytes}, walletNode)
 		if err != nil {

node/node.go

@@ -160,13 +160,12 @@ func (node *Node) getTransactionsForNewBlockAccount(maxNumTxs int) (types.Transa
 }
 
 // StartServer starts a server and process the request by a handler.
-func (node *Node) StartServer(port string) {
+func (node *Node) StartServer() {
 	if p2p.Version == 1 {
-		fmt.Println("going to start server on port:", port)
+		node.log.Debug("Starting server", "node", node, "ip", node.SelfPeer.IP, "port", node.SelfPeer.Port)
-		//node.log.Debug("Starting server", "node", node, "port", port)
+		node.listenOnPort(node.SelfPeer.Port)
-		node.listenOnPort(port)
 	} else {
-		p2pv2.InitHost(node.SelfPeer.IP, port)
+		p2pv2.InitHost(node.SelfPeer.IP, node.SelfPeer.Port)
 		p2pv2.BindHandler(node.NodeHandlerV1)
 		// Hang forever
 		<-make(chan struct{})

node/node_test.go

@@ -6,15 +6,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/harmony-one/harmony/utils"
+	"github.com/harmony-one/harmony/consensus"
-
 	"github.com/harmony-one/harmony/crypto"
 	"github.com/harmony-one/harmony/crypto/pki"
-
-	"github.com/harmony-one/harmony/consensus"
 	"github.com/harmony-one/harmony/p2p"
-
 	proto_node "github.com/harmony-one/harmony/proto/node"
+	"github.com/harmony-one/harmony/utils"
 )
 
 func TestNewNewNode(t *testing.T) {
@@ -181,5 +178,5 @@ func TestPingPongHandler(test *testing.T) {
 	//go sendPingMessage(leader)
 	go sendPongMessage(leader)
 	go exitServer()
-	node.StartServer("8881")
+	node.StartServer()
 }