toolbar/precog/monitor/connection.go

   1 package monitor
   2
   3 import (
   4         "net"
   5         "time"
   6
   7         log "github.com/sirupsen/logrus"
   8         "github.com/vapor/p2p"
   9         "github.com/vapor/toolbar/precog/database/orm"
  10 )
  11
  12 func (m *monitor) connectionRoutine() {
  13         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqMinutes) * time.Minute)
  14         for ; true; <-ticker.C {
  15                 if err := m.dialNodes(); err != nil {
  16                         log.WithFields(log.Fields{"err": err}).Error("dialNodes")
  17                 }
  18         }
  19 }
  20
  21 func (m *monitor) dialNodes() error {
  22         log.Info("Start to reconnect to nodes...")
  23         var nodes []*orm.Node
  24         if err := m.db.Model(&orm.Node{}).Find(&nodes).Error; err != nil {
  25                 return err
  26         }
  27
  28         addresses := make([]*p2p.NetAddress, 0)
  29         for i := 0; i < len(nodes); i++ {
  30                 address := p2p.NewNetAddressIPPort(net.ParseIP(nodes[i].IP), nodes[i].Port)
  31                 addresses = append(addresses, address)
  32         }
  33
  34         // connected peers will be skipped in switch.DialPeers()
  35         m.sw.DialPeers(addresses)
  36         log.Info("DialPeers done.")
  37         peerList := m.sw.GetPeers().List()
  38         m.processDialResults(peerList)
  39         m.checkStatus(peerList)
  40         return nil
  41 }
  42
  43 func (m *monitor) checkStatus(peerList []*p2p.Peer) {
  44         for _, peer := range peerList {
  45                 peer.Start()
  46                 m.peers.AddPeer(peer)
  47         }
  48         log.WithFields(log.Fields{"num": len(m.sw.GetPeers().List()), "peers": m.sw.GetPeers().List()}).Info("connected peers")
  49
  50         for _, peer := range peerList {
  51                 p := m.peers.GetPeer(peer.ID())
  52                 if p == nil {
  53                         continue
  54                 }
  55
  56                 if err := p.SendStatus(m.chain.BestBlockHeader(), m.chain.LastIrreversibleHeader()); err != nil {
  57                         log.WithFields(log.Fields{"peer": p, "err": err}).Error("SendStatus")
  58                         m.peers.RemovePeer(p.ID())
  59                 }
  60         }
  61
  62         for _, peerInfo := range m.peers.GetPeerInfos() {
  63                 if peerInfo.Height > m.bestHeightSeen {
  64                         m.bestHeightSeen = peerInfo.Height
  65                 }
  66         }
  67         log.WithFields(log.Fields{"bestHeight": m.bestHeightSeen}).Info("peersInfo")
  68         m.processPeerInfos(m.peers.GetPeerInfos())
  69
  70         for _, peer := range peerList {
  71                 p := m.peers.GetPeer(peer.ID())
  72                 if p == nil {
  73                         continue
  74                 }
  75
  76                 m.peers.RemovePeer(p.ID())
  77         }
  78         log.Info("Disonnect all peers.")
  79 }