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                 &connection.ChannelDescriptor{
  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         if ok := peer.TrySend(BlockchainChannel, struct{ BlockchainMessage }{&StatusRequestMessage{}}); !ok {
  66                 return errStatusRequest
  67         }
  68
  69         checkTicker := time.NewTimer(handshakeCheckPerid)
  70         timeoutTicker := time.NewTimer(handshakeTimeout)
  71         for {
  72                 select {
  73                 case <-checkTicker.C:
  74                         if exist := pr.peers.getPeer(peer.Key); exist != nil {
  75                                 pr.sm.syncTransactions(peer.Key)
  76                                 return nil
  77                         }
  78
  79                 case <-timeoutTicker.C:
  80                         return errProtocolHandshakeTimeout
  81                 }
  82         }
  83 }
  84
  85 // RemovePeer implements Reactor by removing peer from the pool.
  86 func (pr *ProtocolReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
  87         pr.peers.removePeer(peer.Key)
  88 }
  89
  90 // Receive implements Reactor by handling 4 types of messages (look below).
  91 func (pr *ProtocolReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
  92         msgType, msg, err := DecodeMessage(msgBytes)
  93         if err != nil {
  94                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on reactor decoding message")
  95                 return
  96         }
  97
  98         pr.sm.processMsg(src, msgType, msg)
  99 }