netsync/sync_manager.go

   1 package netsync
   2
   3 import (
   4         "errors"
   5
   6         "github.com/sirupsen/logrus"
   7
   8         "github.com/vapor/config"
   9         "github.com/vapor/consensus"
  10         "github.com/vapor/event"
  11         "github.com/vapor/netsync/chainmgr"
  12         "github.com/vapor/netsync/consensusmgr"
  13         "github.com/vapor/netsync/peers"
  14         "github.com/vapor/p2p"
  15         "github.com/vapor/protocol"
  16 )
  17
  18 const (
  19         logModule = "netsync"
  20 )
  21
  22 var (
  23         errVaultModeDialPeer = errors.New("can't dial peer in vault mode")
  24 )
  25
  26 // ChainMgr is the interface for p2p chain message sync manager.
  27 type ChainMgr interface {
  28         Start() error
  29         IsCaughtUp() bool
  30         Stop()
  31 }
  32
  33 // ConsensusMgr is the interface for consensus message sync manager.
  34 type ConsensusMgr interface {
  35         Start() error
  36         Stop()
  37 }
  38
  39 // Switch is the interface for p2p switch.
  40 type Switch interface {
  41         Start() (bool, error)
  42         Stop() bool
  43         IsListening() bool
  44         DialPeerWithAddress(addr *p2p.NetAddress) error
  45         Peers() *p2p.PeerSet
  46 }
  47
  48 //SyncManager Sync Manager is responsible for the business layer information synchronization
  49 type SyncManager struct {
  50         config       *config.Config
  51         sw           Switch
  52         chainMgr     ChainMgr
  53         consensusMgr ConsensusMgr
  54         peers        *peers.PeerSet
  55 }
  56
  57 // NewSyncManager create sync manager and set switch.
  58 func NewSyncManager(config *config.Config, chain *protocol.Chain, txPool *protocol.TxPool, dispatcher *event.Dispatcher) (*SyncManager, error) {
  59         sw, err := p2p.NewSwitch(config)
  60         if err != nil {
  61                 return nil, err
  62         }
  63         peers := peers.NewPeerSet(sw)
  64
  65         chainManger, err := chainmgr.NewManager(config, sw, chain, txPool, dispatcher, peers)
  66         if err != nil {
  67                 return nil, err
  68         }
  69         consensusMgr := consensusmgr.NewManager(sw, chain, dispatcher, peers)
  70         return &SyncManager{
  71                 config:       config,
  72                 sw:           sw,
  73                 chainMgr:     chainManger,
  74                 consensusMgr: consensusMgr,
  75                 peers:        peers,
  76         }, nil
  77 }
  78
  79 // Start message sync manager service.
  80 func (sm *SyncManager) Start() error {
  81         if _, err := sm.sw.Start(); err != nil {
  82                 logrus.WithFields(logrus.Fields{"module": logModule, "err": err}).Error("failed start switch")
  83                 return err
  84         }
  85
  86         if err := sm.chainMgr.Start(); err != nil {
  87                 return err
  88         }
  89
  90         return sm.consensusMgr.Start()
  91 }
  92
  93 // Stop message sync manager service.
  94 func (sm *SyncManager) Stop() {
  95         sm.chainMgr.Stop()
  96         sm.consensusMgr.Stop()
  97         if !sm.config.VaultMode {
  98                 sm.sw.Stop()
  99         }
 100
 101 }
 102
 103 // IsListening check if the bytomd service port is open?
 104 func (sm *SyncManager) IsListening() bool {
 105         if sm.config.VaultMode {
 106                 return false
 107         }
 108         return sm.sw.IsListening()
 109
 110 }
 111
 112 //IsCaughtUp check wheather the peer finish the sync
 113 func (sm *SyncManager) IsCaughtUp() bool {
 114         return sm.chainMgr.IsCaughtUp()
 115 }
 116
 117 // PeerCount count the number of connected peers.
 118 func (sm *SyncManager) PeerCount() int {
 119         if sm.config.VaultMode {
 120                 return 0
 121         }
 122         return len(sm.sw.Peers().List())
 123 }
 124
 125 // GetNetwork get the type of network.
 126 func (sm *SyncManager) GetNetwork() string {
 127         return sm.config.ChainID
 128 }
 129
 130 // BestPeer fine the peer with the highest height from the connected peers.
 131 func (sm *SyncManager) BestPeer() *peers.PeerInfo {
 132         bestPeer := sm.peers.BestPeer(consensus.SFFullNode)
 133         if bestPeer != nil {
 134                 return bestPeer.GetPeerInfo()
 135         }
 136         return nil
 137 }
 138
 139 // DialPeerWithAddress dial the peer and establish a connection.
 140 func (sm *SyncManager) DialPeerWithAddress(addr *p2p.NetAddress) error {
 141         if sm.config.VaultMode {
 142                 return errVaultModeDialPeer
 143         }
 144
 145         return sm.sw.DialPeerWithAddress(addr)
 146 }
 147
 148 //GetPeerInfos return peer info of all connected peers.
 149 func (sm *SyncManager) GetPeerInfos() []*peers.PeerInfo {
 150         return sm.peers.GetPeerInfos()
 151 }
 152
 153 //StopPeer try to stop peer by given ID
 154 func (sm *SyncManager) StopPeer(peerID string) error {
 155         if peer := sm.peers.GetPeer(peerID); peer == nil {
 156                 return errors.New("peerId not exist")
 157         }
 158         sm.peers.RemovePeer(peerID)
 159         return nil
 160 }