toolbar/precognitive/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         "fmt"
   5         "os"
   6         "os/user"
   7         "path"
   8         "strings"
   9
  10         "github.com/jinzhu/gorm"
  11         log "github.com/sirupsen/logrus"
  12
  13         vaporCfg "github.com/bytom/vapor/config"
  14         "github.com/bytom/vapor/crypto/ed25519/chainkd"
  15         dbm "github.com/bytom/vapor/database/leveldb"
  16         "github.com/bytom/vapor/event"
  17         "github.com/bytom/vapor/netsync/chainmgr"
  18         "github.com/bytom/vapor/netsync/consensusmgr"
  19         "github.com/bytom/vapor/netsync/peers"
  20         "github.com/bytom/vapor/p2p"
  21         "github.com/bytom/vapor/p2p/discover/dht"
  22         "github.com/bytom/vapor/p2p/discover/mdns"
  23         "github.com/bytom/vapor/p2p/signlib"
  24         "github.com/bytom/vapor/test/mock"
  25         "github.com/bytom/vapor/toolbar/precognitive/config"
  26 )
  27
  28 type monitor struct {
  29         cfg            *config.Config
  30         db             *gorm.DB
  31         nodeCfg        *vaporCfg.Config
  32         sw             *p2p.Switch
  33         privKey        chainkd.XPrv
  34         chain          *mock.Chain
  35         txPool         *mock.Mempool
  36         bestHeightSeen uint64
  37         peers          *peers.PeerSet
  38 }
  39
  40 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  41         dbPath, err := makePath()
  42         if err != nil {
  43                 log.WithFields(log.Fields{"err": err}).Fatal("makePath")
  44         }
  45
  46         nodeCfg := &vaporCfg.Config{
  47                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  48                 P2P:        vaporCfg.DefaultP2PConfig(),
  49                 Federation: vaporCfg.DefaultFederationConfig(),
  50         }
  51         nodeCfg.DBPath = dbPath
  52         nodeCfg.ChainID = "mainnet"
  53         privKey, err := signlib.NewPrivKey()
  54         if err != nil {
  55                 log.WithFields(log.Fields{"err": err}).Fatal("NewPrivKey")
  56         }
  57
  58         chain, txPool, err := mockChainAndPool()
  59         if err != nil {
  60                 log.WithFields(log.Fields{"err": err}).Fatal("mockChainAndPool")
  61         }
  62
  63         return &monitor{
  64                 cfg:            cfg,
  65                 db:             db,
  66                 nodeCfg:        nodeCfg,
  67                 privKey:        privKey.(chainkd.XPrv),
  68                 chain:          chain,
  69                 txPool:         txPool,
  70                 bestHeightSeen: uint64(0),
  71         }
  72 }
  73
  74 func makePath() (string, error) {
  75         usr, err := user.Current()
  76         if err != nil {
  77                 return "", err
  78         }
  79
  80         dataPath := path.Join(usr.HomeDir, "/.vapor_precog")
  81         if err := os.MkdirAll(dataPath, os.ModePerm); err != nil {
  82                 return "", err
  83         }
  84
  85         return dataPath, nil
  86 }
  87
  88 func (m *monitor) Run() {
  89         if err := m.makeSwitch(); err != nil {
  90                 log.WithFields(log.Fields{"err": err}).Fatal("makeSwitch")
  91         }
  92
  93         go m.discoveryRoutine()
  94         go m.connectionRoutine()
  95 }
  96
  97 func (m *monitor) makeSwitch() error {
  98         var seeds []string
  99         for _, node := range m.cfg.Nodes {
 100                 seeds = append(seeds, fmt.Sprintf("%s:%d", node.IP, node.Port))
 101         }
 102         m.nodeCfg.P2P.Seeds = strings.Join(seeds, ",")
 103
 104         l, listenAddr := p2p.GetListener(m.nodeCfg.P2P)
 105         discv, err := dht.NewDiscover(m.nodeCfg, m.privKey, l.ExternalAddress().Port, m.cfg.NetworkID)
 106         if err != nil {
 107                 return err
 108         }
 109
 110         // no need for lanDiscv, but passing &mdns.LANDiscover{} will cause NilPointer
 111         lanDiscv := mdns.NewLANDiscover(mdns.NewProtocol(m.nodeCfg.ChainID), int(l.ExternalAddress().Port))
 112         m.sw, err = p2p.NewSwitch(m.nodeCfg, discv, lanDiscv, l, m.privKey, listenAddr, m.cfg.NetworkID)
 113         if err != nil {
 114                 return err
 115         }
 116
 117         m.peers = peers.NewPeerSet(m.sw)
 118         return m.prepareReactors()
 119 }
 120
 121 func (m *monitor) prepareReactors() error {
 122         dispatcher := event.NewDispatcher()
 123         // add ConsensusReactor for consensusChannel
 124         _ = consensusmgr.NewManager(m.sw, m.chain, m.peers, dispatcher)
 125         fastSyncDB := dbm.NewDB("fastsync", m.nodeCfg.DBBackend, m.nodeCfg.DBDir())
 126         // add ProtocolReactor to handle msgs
 127         if _, err := chainmgr.NewManager(m.nodeCfg, m.sw, m.chain, m.txPool, dispatcher, m.peers, fastSyncDB); err != nil {
 128                 return err
 129         }
 130
 131         for label, reactor := range m.sw.GetReactors() {
 132                 log.WithFields(log.Fields{"label": label, "reactor": reactor}).Debug("start reactor")
 133                 if _, err := reactor.Start(); err != nil {
 134                         return err
 135                 }
 136         }
 137
 138         m.sw.GetSecurity().RegisterFilter(m.sw.GetNodeInfo())
 139         m.sw.GetSecurity().RegisterFilter(m.sw.GetPeers())
 140         return m.sw.GetSecurity().Start()
 141 }