func (conR *ConsensusReactor) AddPeer(peer p2p.Peer) {
	if !conR.IsRunning() {
		return
	}
	peerState := NewPeerState(peer).SetLogger(conR.Logger)
	peer.Set(types.PeerStateKey, peerState)
	go conR.gossipDataRoutine(peer, peerState)
	go conR.gossipVotesRoutine(peer, peerState)
	go conR.queryMaj23Routine(peer, peerState)
	if !conR.FastSync() {
		conR.sendNewRoundStepMessage(peer)
	}
}
AddPeer implements Reactor
package consensus
import (
	"fmt"
	"reflect"
	"sync"
	"time"
	"github.com/pkg/errors"
	amino "github.com/tendermint/go-amino"
	cstypes "github.com/XuanMaoSecLab/DolphinChain/consensus/types"
	cmn "github.com/XuanMaoSecLab/DolphinChain/libs/common"
	tmevents "github.com/XuanMaoSecLab/DolphinChain/libs/events"
	"github.com/XuanMaoSecLab/DolphinChain/libs/log"
	"github.com/XuanMaoSecLab/DolphinChain/p2p"
	sm "github.com/XuanMaoSecLab/DolphinChain/state"
	"github.com/XuanMaoSecLab/DolphinChain/types"
	tmtime "github.com/XuanMaoSecLab/DolphinChain/types/time"
)
const (
	StateChannel       = byte(0x20)
	DataChannel        = byte(0x21)
	VoteChannel        = byte(0x22)
	VoteSetBitsChannel = byte(0x23)
	maxMsgSize = 1048576  
	blocksToContributeToBecomeGoodPeer = 10000
	votesToContributeToBecomeGoodPeer  = 10000
)
type ConsensusReactor struct {
	p2p.BaseReactor  
	conS *ConsensusState
	mtx      sync.RWMutex
	fastSync bool
	eventBus *types.EventBus
	metrics *Metrics
}
type ReactorOption func(*ConsensusReactor)
func NewConsensusReactor(consensusState *ConsensusState, fastSync bool, options ...ReactorOption) *ConsensusReactor {
	conR := &ConsensusReactor{
		conS:     consensusState,
		fastSync: fastSync,
		metrics:  NopMetrics(),
	}
	conR.updateFastSyncingMetric()
	conR.BaseReactor = *p2p.NewBaseReactor("ConsensusReactor", conR)
	for _, option := range options {
		option(conR)
	}
	return conR
}
func (conR *ConsensusReactor) OnStart() error {
	conR.Logger.Info("ConsensusReactor ", "fastSync", conR.FastSync())
	go conR.peerStatsRoutine()
	conR.subscribeToBroadcastEvents()
	if !conR.FastSync() {
		err := conR.conS.Start()
		if err != nil {
			return err
		}
	}
	return nil
}
func (conR *ConsensusReactor) OnStop() {
	conR.unsubscribeFromBroadcastEvents()
	conR.conS.Stop()
	if !conR.FastSync() {
		conR.conS.Wait()
	}
}
func (conR *ConsensusReactor) SwitchToConsensus(state sm.State, blocksSynced int) {
	conR.Logger.Info("SwitchToConsensus")
	conR.conS.reconstructLastCommit(state)
	conR.conS.updateToState(state)
	conR.mtx.Lock()
	conR.fastSync = false
	conR.mtx.Unlock()
	conR.metrics.FastSyncing.Set(0)
	if blocksSynced > 0 {
		conR.conS.doWALCatchup = false
	}
	err := conR.conS.Start()
	if err != nil {
		conR.Logger.Error("Error starting conS", "err", err)
		return
	}
}
func (conR *ConsensusReactor) GetChannels() []*p2p.ChannelDescriptor {
	return []*p2p.ChannelDescriptor{
		{
			ID:                  StateChannel,
			Priority:            5,
			SendQueueCapacity:   100,
			RecvMessageCapacity: maxMsgSize,
		},
		{
			ID:                  DataChannel,  
			Priority:            10,           
			SendQueueCapacity:   100,
			RecvBufferCapacity:  50 * 4096,
			RecvMessageCapacity: maxMsgSize,
		},
		{
			ID:                  VoteChannel,
			Priority:            5,
			SendQueueCapacity:   100,
			RecvBufferCapacity:  100 * 100,
			RecvMessageCapacity: maxMsgSize,
		},
		{
			ID:                  VoteSetBitsChannel,
			Priority:            1,
			SendQueueCapacity:   2,
			RecvBufferCapacity:  1024,
			RecvMessageCapacity: maxMsgSize,
		},
	}
}