Use Network in Consensus core (#85)

5 files changed, 163 insertions, 3 deletions

diff --git a/core/consensus.go b/core/consensus.go
index 9c109e4..d6b5efd 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -19,6 +19,7 @@ package core
 
 import (
 	"fmt"
+	"sort"
 	"sync"
 	"time"
 
@@ -56,6 +57,7 @@ var (
 
 // Consensus implements DEXON Consensus algorithm.
 type Consensus struct {
+	ID       types.ValidatorID
 	app      Application
 	gov      Governance
 	rbModule *reliableBroadcast
@@ -63,9 +65,12 @@ type Consensus struct {
 	ctModule *consensusTimestamp
 	ccModule *compactionChain
 	db       blockdb.BlockDatabase
+	network  Network
+	tick     *time.Ticker
 	prvKey   crypto.PrivateKey
 	sigToPub SigToPubFn
 	lock     sync.RWMutex
+	stopChan chan struct{}
 }
 
 // NewConsensus construct an Consensus instance.
@@ -73,6 +78,8 @@ func NewConsensus(
 	app Application,
 	gov Governance,
 	db blockdb.BlockDatabase,
+	network Network,
+	tick *time.Ticker,
 	prv crypto.PrivateKey,
 	sigToPub SigToPubFn) *Consensus {
 	validatorSet := gov.GetValidatorSet()
@@ -95,6 +102,7 @@ func NewConsensus(
 		validators)
 
 	return &Consensus{
+		ID:       types.NewValidatorID(prv.PublicKey()),
 		rbModule: rb,
 		toModule: to,
 		ctModule: newConsensusTimestamp(),
@@ -102,8 +110,94 @@ func NewConsensus(
 		app:      newNonBlockingApplication(app),
 		gov:      gov,
 		db:       db,
+		network:  network,
+		tick:     tick,
 		prvKey:   prv,
 		sigToPub: sigToPub,
+		stopChan: make(chan struct{}),
+	}
+}
+
+// Run starts running Consensus core.
+func (con *Consensus) Run() {
+	go con.processMsg(con.network.ReceiveChan())
+
+	chainID := uint64(0)
+	hashes := make(common.Hashes, 0, len(con.gov.GetValidatorSet()))
+	for vID := range con.gov.GetValidatorSet() {
+		hashes = append(hashes, vID.Hash)
+	}
+	sort.Sort(hashes)
+	for i, hash := range hashes {
+		if hash == con.ID.Hash {
+			chainID = uint64(i)
+			break
+		}
+	}
+
+	genesisBlock := &types.Block{
+		ProposerID: con.ID,
+		ChainID:    chainID,
+	}
+	if err := con.PrepareGenesisBlock(genesisBlock, time.Now().UTC()); err != nil {
+		fmt.Println(err)
+	}
+	if err := con.ProcessBlock(genesisBlock); err != nil {
+		fmt.Println(err)
+	}
+	con.network.BroadcastBlock(genesisBlock)
+
+ProposingBlockLoop:
+	for {
+		select {
+		case <-con.tick.C:
+		case <-con.stopChan:
+			break ProposingBlockLoop
+		}
+		block := &types.Block{
+			ProposerID: con.ID,
+			ChainID:    chainID,
+		}
+		if err := con.PrepareBlock(block, time.Now().UTC()); err != nil {
+			fmt.Println(err)
+		}
+		if err := con.ProcessBlock(block); err != nil {
+			fmt.Println(err)
+		}
+		con.network.BroadcastBlock(block)
+	}
+}
+
+// Stop the Consensus core.
+func (con *Consensus) Stop() {
+	con.stopChan <- struct{}{}
+	con.stopChan <- struct{}{}
+}
+
+func (con *Consensus) processMsg(msgChan <-chan interface{}) {
+	for {
+		var msg interface{}
+		select {
+		case msg = <-msgChan:
+		case <-con.stopChan:
+			return
+		}
+
+		switch val := msg.(type) {
+		case *types.Block:
+			if err := con.ProcessBlock(val); err != nil {
+				fmt.Println(err)
+			}
+			types.RecycleBlock(val)
+		case *types.NotaryAck:
+			if err := con.ProcessNotaryAck(val); err != nil {
+				fmt.Println(err)
+			}
+		case *types.Vote:
+			if err := con.ProcessVote(val); err != nil {
+				fmt.Println(err)
+			}
+		}
 	}
 }
 
diff --git a/core/consensus_test.go b/core/consensus_test.go
index 46df8eb..70ab5aa 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -30,6 +30,26 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
+// network implements core.Network.
+type network struct {
+}
+
+// BroadcastVote broadcasts vote to all nodes in DEXON network.
+func (n *network) BroadcastVote(vote *types.Vote) {}
+
+// BroadcastBlock broadcasts block to all nodes in DEXON network.
+func (n *network) BroadcastBlock(block *types.Block) {
+}
+
+// BroadcastNotaryAck broadcasts notaryAck to all nodes in DEXON network.
+func (n *network) BroadcastNotaryAck(notaryAck *types.NotaryAck) {
+}
+
+// ReceiveChan returns a channel to receive messages from DEXON network.
+func (n *network) ReceiveChan() <-chan interface{} {
+	return make(chan interface{})
+}
+
 type ConsensusTestSuite struct {
 	suite.Suite
 }
@@ -56,7 +76,8 @@ func (s *ConsensusTestSuite) prepareConsensus(
 	s.Require().Nil(err)
 	prv, exist := gov.PrivateKeys[vID]
 	s.Require().True(exist)
-	con := NewConsensus(app, gov, db, prv, eth.SigToPub)
+	con := NewConsensus(app, gov, db,
+		&network{}, time.NewTicker(1), prv, eth.SigToPub)
 	return &con.app, con
 }
 
diff --git a/integration_test/network.go b/integration_test/network.go
new file mode 100644
index 0000000..446d01a
--- /dev/null
+++ b/integration_test/network.go
@@ -0,0 +1,42 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus-core library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package integration
+
+import (
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+)
+
+// Network implements core.Network.
+type Network struct {
+}
+
+// BroadcastVote broadcasts vote to all nodes in DEXON network.
+func (n *Network) BroadcastVote(vote *types.Vote) {}
+
+// BroadcastBlock broadcasts block to all nodes in DEXON network.
+func (n *Network) BroadcastBlock(block *types.Block) {
+}
+
+// BroadcastNotaryAck broadcasts notaryAck to all nodes in DEXON network.
+func (n *Network) BroadcastNotaryAck(notaryAck *types.NotaryAck) {
+}
+
+// ReceiveChan returns a channel to receive messages from DEXON network.
+func (n *Network) ReceiveChan() <-chan interface{} {
+	return make(chan interface{})
+}
diff --git a/integration_test/validator.go b/integration_test/validator.go
index 60c6b1f..53f9456 100644
--- a/integration_test/validator.go
+++ b/integration_test/validator.go
@@ -102,7 +102,7 @@ func NewValidator(
 		networkLatency:   networkLatency,
 		proposingLatency: proposingLatency,
 		cons: core.NewConsensus(
-			app, gov, db, privateKey, eth.SigToPub),
+			app, gov, db, &Network{}, time.NewTicker(1), privateKey, eth.SigToPub),
 	}
 }
 
diff --git a/simulation/validator.go b/simulation/validator.go
index 8a672c5..a54c848 100644
--- a/simulation/validator.go
+++ b/simulation/validator.go
@@ -100,7 +100,10 @@ func (v *Validator) Run() {
 		}
 	}
 	v.consensus = core.NewConsensus(
-		v.app, v.gov, v.db, v.prvKey, v.sigToPub)
+		v.app, v.gov, v.db, v.network,
+		time.NewTicker(
+			time.Duration(v.config.ProposeIntervalMean)*time.Millisecond),
+		v.prvKey, v.sigToPub)
 
 	genesisBlock := &types.Block{
 		ProposerID: v.ID,