func (r *Reactor) OnStart() error {
	r.Logger.Debug("consensus wait sync", "wait_sync", r.WaitSync())
	go r.peerStatsRoutine()
	r.subscribeToBroadcastEvents()
	if !r.WaitSync() {
		if err := r.state.Start(); err != nil {
			return err
		}
	}
	go r.processStateCh()
	go r.processDataCh()
	go r.processVoteCh()
	go r.processVoteSetBitsCh()
	go r.processPeerUpdates()
	return nil
}
OnStart starts separate go routines for each p2p Channel and listens for
envelopes on each. In addition, it also listens for peer updates and handles
messages on that p2p channel accordingly. The caller must be sure to execute
OnStop to ensure the outbound p2p Channels are closed.
package consensus
import (
	"fmt"
	"runtime/debug"
	"time"
	cstypes "github.com/tendermint/tendermint/internal/consensus/types"
	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
	"github.com/tendermint/tendermint/internal/p2p"
	sm "github.com/tendermint/tendermint/internal/state"
	"github.com/tendermint/tendermint/libs/bits"
	tmevents "github.com/tendermint/tendermint/libs/events"
	"github.com/tendermint/tendermint/libs/log"
	"github.com/tendermint/tendermint/libs/service"
	tmcons "github.com/tendermint/tendermint/proto/tendermint/consensus"
	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
	"github.com/tendermint/tendermint/types"
)
var (
	_ service.Service = (*Reactor)(nil)
	_ p2p.Wrapper     = (*tmcons.Message)(nil)
)
func GetChannelDescriptors() []*p2p.ChannelDescriptor {
	return []*p2p.ChannelDescriptor{
		{
			ID:                  StateChannel,
			MessageType:         new(tmcons.Message),
			Priority:            8,
			SendQueueCapacity:   64,
			RecvMessageCapacity: maxMsgSize,
			RecvBufferCapacity:  128,
		},
		{
			ID:                  DataChannel,
			MessageType:         new(tmcons.Message),
			Priority:            12,
			SendQueueCapacity:   64,
			RecvBufferCapacity:  512,
			RecvMessageCapacity: maxMsgSize,
		},
		{
			ID:                  VoteChannel,
			MessageType:         new(tmcons.Message),
			Priority:            10,
			SendQueueCapacity:   64,
			RecvBufferCapacity:  128,
			RecvMessageCapacity: maxMsgSize,
		},
		{
			ID:                  VoteSetBitsChannel,
			MessageType:         new(tmcons.Message),
			Priority:            5,
			SendQueueCapacity:   8,
			RecvBufferCapacity:  128,
			RecvMessageCapacity: maxMsgSize,
		},
	}
}
const (
	StateChannel       = p2p.ChannelID(0x20)
	DataChannel        = p2p.ChannelID(0x21)
	VoteChannel        = p2p.ChannelID(0x22)
	VoteSetBitsChannel = p2p.ChannelID(0x23)
	maxMsgSize = 1048576  
	blocksToContributeToBecomeGoodPeer = 10000
	votesToContributeToBecomeGoodPeer  = 10000
	listenerIDConsensus = "consensus-reactor"
)
type ReactorOption func(*Reactor)
type BlockSyncReactor interface {
	SwitchToBlockSync(sm.State) error
	GetMaxPeerBlockHeight() int64
	GetTotalSyncedTime() time.Duration
	GetRemainingSyncTime() time.Duration
}
type ConsSyncReactor interface {
	SwitchToConsensus(sm.State, bool)
	SetStateSyncingMetrics(float64)
	SetBlockSyncingMetrics(float64)
}
type Reactor struct {
	service.BaseService
	state    *State
	eventBus *types.EventBus
	Metrics  *Metrics
	mtx      tmsync.RWMutex
	peers    map[types.NodeID]*PeerState
	waitSync bool
	stateCh       *p2p.Channel
	dataCh        *p2p.Channel
	voteCh        *p2p.Channel
	voteSetBitsCh *p2p.Channel
	peerUpdates   *p2p.PeerUpdates
	stateCloseCh chan struct{}
	closeCh      chan struct{}
}
func NewReactor(
	logger log.Logger,
	cs *State,
	stateCh *p2p.Channel,
	dataCh *p2p.Channel,
	voteCh *p2p.Channel,
	voteSetBitsCh *p2p.Channel,
	peerUpdates *p2p.PeerUpdates,
	waitSync bool,
	options ...ReactorOption,
) *Reactor {
	r := &Reactor{
		state:         cs,
		waitSync:      waitSync,
		peers:         make(map[types.NodeID]*PeerState),
		Metrics:       NopMetrics(),
		stateCh:       stateCh,
		dataCh:        dataCh,
		voteCh:        voteCh,
		voteSetBitsCh: voteSetBitsCh,
		peerUpdates:   peerUpdates,
		stateCloseCh:  make(chan struct{}),
		closeCh:       make(chan struct{}),
	}
	r.BaseService = *service.NewBaseService(logger, "Consensus", r)
	for _, opt := range options {
		opt(r)
	}
	return r
}