netsync/consensus/reactor.go

   1 package consensus
   2
   3 import (
   4         log "github.com/sirupsen/logrus"
   5
   6         "github.com/vapor/p2p"
   7         "github.com/vapor/p2p/connection"
   8 )
   9
  10 const (
  11         logModule                 = "consensus"
  12         ConsensusChannel          = byte(0x50)
  13         maxBlockchainResponseSize = 22020096 + 2
  14 )
  15
  16 type ConsensusReactor struct {
  17         p2p.BaseReactor
  18         mgr *Manager
  19 }
  20
  21 func NewConsensusReactor(manager *Manager) *ConsensusReactor {
  22         cr := &ConsensusReactor{
  23                 mgr: manager,
  24         }
  25         cr.BaseReactor = *p2p.NewBaseReactor("ConsensusReactor", cr)
  26         return cr
  27 }
  28
  29 // GetChannels implements Reactor
  30 func (cr *ConsensusReactor) GetChannels() []*connection.ChannelDescriptor {
  31         return []*connection.ChannelDescriptor{
  32                 {
  33                         ID:                ConsensusChannel,
  34                         Priority:          10,
  35                         SendQueueCapacity: 100,
  36                 },
  37         }
  38 }
  39
  40 // OnStart implements BaseService
  41 func (cr *ConsensusReactor) OnStart() error {
  42         cr.BaseReactor.OnStart()
  43         return nil
  44 }
  45
  46 // OnStop implements BaseService
  47 func (cr *ConsensusReactor) OnStop() {
  48         cr.BaseReactor.OnStop()
  49 }
  50
  51 // AddPeer implements Reactor by sending our state to peer.
  52 func (cr *ConsensusReactor) AddPeer(peer *p2p.Peer) error {
  53         cr.mgr.AddPeer(peer)
  54         return nil
  55 }
  56
  57 // RemovePeer implements Reactor by removing peer from the pool.
  58 func (cr *ConsensusReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
  59         cr.mgr.RemovePeer(peer.Key)
  60 }
  61
  62 // Receive implements Reactor by handling 4 types of messages (look below).
  63 func (cr *ConsensusReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
  64         msgType, msg, err := decodeMessage(msgBytes)
  65         if err != nil {
  66                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on reactor decoding message")
  67                 return
  68         }
  69
  70         cr.mgr.processMsg(src.ID(), msgType, msg)
  71 }