netsync/protocol_reactor.go

   1 package netsync
   2
   3 import (
   4         "time"
   5
   6         log "github.com/sirupsen/logrus"
   7
   8         "github.com/vapor/errors"
   9         "github.com/vapor/p2p"
  10         "github.com/vapor/p2p/connection"
  11 )
  12
  13 const (
  14         handshakeTimeout    = 10 * time.Second
  15         handshakeCheckPerid = 500 * time.Millisecond
  16 )
  17
  18 var (
  19         errProtocolHandshakeTimeout = errors.New("Protocol handshake timeout")
  20         errStatusRequest            = errors.New("Status request error")
  21 )
  22
  23 //ProtocolReactor handles new coming protocol message.
  24 type ProtocolReactor struct {
  25         p2p.BaseReactor
  26
  27         sm    *SyncManager
  28         peers *peerSet
  29 }
  30
  31 // NewProtocolReactor returns the reactor of whole blockchain.
  32 func NewProtocolReactor(sm *SyncManager, peers *peerSet) *ProtocolReactor {
  33         pr := &ProtocolReactor{
  34                 sm:    sm,
  35                 peers: peers,
  36         }
  37         pr.BaseReactor = *p2p.NewBaseReactor("ProtocolReactor", pr)
  38         return pr
  39 }
  40
  41 // GetChannels implements Reactor
  42 func (pr *ProtocolReactor) GetChannels() []*connection.ChannelDescriptor {
  43         return []*connection.ChannelDescriptor{
  44                 {
  45                         ID:                BlockchainChannel,
  46                         Priority:          5,
  47                         SendQueueCapacity: 100,
  48                 },
  49         }
  50 }
  51
  52 // OnStart implements BaseService
  53 func (pr *ProtocolReactor) OnStart() error {
  54         pr.BaseReactor.OnStart()
  55         return nil
  56 }
  57
  58 // OnStop implements BaseService
  59 func (pr *ProtocolReactor) OnStop() {
  60         pr.BaseReactor.OnStop()
  61 }
  62
  63 // AddPeer implements Reactor by sending our state to peer.
  64 func (pr *ProtocolReactor) AddPeer(peer *p2p.Peer) error {
  65         pr.sm.AddPeer(peer)
  66         if err := pr.sm.SendStatus(peer); err != nil {
  67                 return err
  68         }
  69         pr.sm.syncTransactions(peer.Key)
  70         return nil
  71 }
  72
  73 // RemovePeer implements Reactor by removing peer from the pool.
  74 func (pr *ProtocolReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
  75         pr.peers.removePeer(peer.Key)
  76 }
  77
  78 // Receive implements Reactor by handling 4 types of messages (look below).
  79 func (pr *ProtocolReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
  80         msgType, msg, err := DecodeMessage(msgBytes)
  81         if err != nil {
  82                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on reactor decoding message")
  83                 return
  84         }
  85
  86         pr.sm.processMsg(src, msgType, msg)
  87 }