go-ethereum/eth/bft/bft_handler.go

package bft

import (
	"github.com/XinFinOrg/XDPoSChain/consensus"
	"github.com/XinFinOrg/XDPoSChain/consensus/XDPoS"
	"github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/utils"
	"github.com/XinFinOrg/XDPoSChain/core"
	"github.com/XinFinOrg/XDPoSChain/log"
)

//Define Boradcast Group functions
type broadcastVoteFn func(*utils.Vote)
type broadcastTimeoutFn func(*utils.Timeout)
type broadcastSyncInfoFn func(*utils.SyncInfo)

type Bfter struct {
	blockChainReader consensus.ChainReader
	broadcastCh      chan interface{}
	quit             chan struct{}
	consensus        ConsensusFns
	broadcast        BroadcastFns
}

type ConsensusFns struct {
	verifyVote  func(consensus.ChainReader, *utils.Vote) (bool, error)
	voteHandler func(consensus.ChainReader, *utils.Vote) error

	verifyTimeout  func(consensus.ChainReader, *utils.Timeout) (bool, error)
	timeoutHandler func(consensus.ChainReader, *utils.Timeout) error

	verifySyncInfo  func(consensus.ChainReader, *utils.SyncInfo) (bool, error)
	syncInfoHandler func(consensus.ChainReader, *utils.SyncInfo) error
}

type BroadcastFns struct {
	Vote     broadcastVoteFn
	Timeout  broadcastTimeoutFn
	SyncInfo broadcastSyncInfoFn
}

func New(broadcasts BroadcastFns, blockChainReader *core.BlockChain) *Bfter {

	return &Bfter{
		quit:             make(chan struct{}),
		broadcastCh:      make(chan interface{}),
		broadcast:        broadcasts,
		blockChainReader: blockChainReader,
	}
}

func (b *Bfter) SetConsensusFuns(engine consensus.Engine) {
	e := engine.(*XDPoS.XDPoS)
	b.broadcastCh = e.EngineV2.BroadcastCh
	b.consensus = ConsensusFns{
		verifySyncInfo: e.EngineV2.VerifySyncInfoMessage,
		verifyVote:     e.EngineV2.VerifyVoteMessage,
		verifyTimeout:  e.EngineV2.VerifyTimeoutMessage,

		voteHandler:     e.EngineV2.VoteHandler,
		timeoutHandler:  e.EngineV2.TimeoutHandler,
		syncInfoHandler: e.EngineV2.SyncInfoHandler,
	}
}

func (b *Bfter) Vote(vote *utils.Vote) error {
	log.Trace("Receive Vote", "hash", vote.Hash().Hex(), "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round)

	verified, err := b.consensus.verifyVote(b.blockChainReader, vote)

	if err != nil {
		log.Error("Verify BFT Vote", "error", err)
		return err
	}

	b.broadcastCh <- vote

	if verified {
		err = b.consensus.voteHandler(b.blockChainReader, vote)
		if err != nil {
			if _, ok := err.(*utils.ErrIncomingMessageRoundTooFarFromCurrentRound); ok {
				log.Warn("vote round not equal", "error", err, "vote", vote.Hash())
				return err
			}
			log.Error("handle BFT Vote", "error", err)
			return err
		}
	}

	return nil
}
func (b *Bfter) Timeout(timeout *utils.Timeout) error {
	log.Debug("Receive Timeout", "timeout", timeout)

	verified, err := b.consensus.verifyTimeout(b.blockChainReader, timeout)
	if err != nil {
		log.Error("Verify BFT Timeout", "timeoutRound", timeout.Round, "timeoutGapNum", timeout.GapNumber, "error", err)
		return err
	}

	b.broadcastCh <- timeout
	if verified {
		err = b.consensus.timeoutHandler(b.blockChainReader, timeout)
		if err != nil {
			if _, ok := err.(*utils.ErrIncomingMessageRoundNotEqualCurrentRound); ok {
				log.Warn("timeout round not equal", "error", err)
				return err
			}
			log.Error("handle BFT Timeout", "error", err)
			return err
		}
	}

	return nil
}
func (b *Bfter) SyncInfo(syncInfo *utils.SyncInfo) error {
	log.Debug("Receive SyncInfo", "syncInfo", syncInfo)

	verified, err := b.consensus.verifySyncInfo(b.blockChainReader, syncInfo)
	if err != nil {
		log.Error("Verify BFT SyncInfo", "error", err)
		return err
	}

	b.broadcastCh <- syncInfo
	// Process only if verified and qualified
	if verified {
		err = b.consensus.syncInfoHandler(b.blockChainReader, syncInfo)
		if err != nil {
			log.Error("handle BFT SyncInfo", "error", err)
			return err
		}
	}
	return nil
}

// Start Bft receiver
func (b *Bfter) Start() {
	go b.loop()
}
func (b *Bfter) Stop() {
	close(b.quit)
}
func (b *Bfter) loop() {
	for {
		select {
		case <-b.quit:
			return
		case obj := <-b.broadcastCh:
			switch v := obj.(type) {
			case *utils.Vote:
				go b.broadcast.Vote(v)
			case *utils.Timeout:
				go b.broadcast.Timeout(v)
			case *utils.SyncInfo:
				go b.broadcast.SyncInfo(v)
			default:
				log.Error("Unknown message type received", "value", v)
			}
		}
	}
}