13 "github.com/jinzhu/gorm"
14 log "github.com/sirupsen/logrus"
15 // dbm "github.com/vapor/database/leveldb"
17 vaporCfg "github.com/vapor/config"
18 "github.com/vapor/crypto/ed25519/chainkd"
19 "github.com/vapor/p2p"
20 // conn "github.com/vapor/p2p/connection"
21 // "github.com/vapor/consensus"
22 // "github.com/vapor/crypto/sha3pool"
23 "github.com/vapor/p2p/discover/dht"
24 "github.com/vapor/p2p/discover/mdns"
25 "github.com/vapor/p2p/signlib"
26 "github.com/vapor/toolbar/precog/config"
27 "github.com/vapor/toolbar/precog/database/orm"
38 nodeCfg *vaporCfg.Config
40 discvCh chan *dht.Node
44 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
45 nodeCfg := &vaporCfg.Config{
46 BaseConfig: vaporCfg.DefaultBaseConfig(),
47 P2P: vaporCfg.DefaultP2PConfig(),
48 Federation: vaporCfg.DefaultFederationConfig(),
50 nodeCfg.DBPath = "vapor_precog_data"
51 nodeCfg.ChainID = "mainnet"
52 discvCh := make(chan *dht.Node)
53 privKey, err := signlib.NewPrivKey()
63 privKey: privKey.(chainkd.XPrv),
67 func (m *monitor) Run() {
68 defer os.RemoveAll(m.nodeCfg.DBPath)
71 for _, node := range m.cfg.Nodes {
72 seeds = append(seeds, fmt.Sprintf("%s:%d", node.Host, node.Port))
75 m.nodeCfg.P2P.Seeds = strings.Join(seeds, ",")
77 sw, err := m.makeSwitch()
87 ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
88 for ; true; <-ticker.C {
94 // create or update: https://github.com/jinzhu/gorm/issues/1307
95 func (m *monitor) upSertNode(node *config.Node) error {
97 node.PublicKey = fmt.Sprintf("%v", node.XPub.PublicKey().String())
100 ormNode := &orm.Node{PublicKey: node.PublicKey}
101 if err := m.db.Where(&orm.Node{PublicKey: node.PublicKey}).First(ormNode).Error; err != nil && err != gorm.ErrRecordNotFound {
105 if node.Alias != "" {
106 ormNode.Alias = node.Alias
108 if node.XPub != nil {
109 ormNode.Xpub = node.XPub.String()
111 ormNode.Host = node.Host
112 ormNode.Port = node.Port
113 return m.db.Where(&orm.Node{PublicKey: ormNode.PublicKey}).
116 Alias: ormNode.Alias,
119 }).FirstOrCreate(ormNode).Error
122 func (m *monitor) makeSwitch() (*p2p.Switch, error) {
123 l, listenAddr := p2p.GetListener(m.nodeCfg.P2P)
124 discv, err := dht.NewDiscover(m.nodeCfg, m.privKey, l.ExternalAddress().Port, m.cfg.NetworkID)
129 // no need for lanDiscv, but passing &mdns.LANDiscover{} will cause NilPointer
130 lanDiscv := mdns.NewLANDiscover(mdns.NewProtocol(), int(l.ExternalAddress().Port))
131 return p2p.NewSwitch(m.nodeCfg, discv, lanDiscv, l, m.privKey, listenAddr, m.cfg.NetworkID)
134 func (m *monitor) discovery() {
135 ticker := time.NewTicker(time.Duration(discvFreqSec) * time.Second)
137 nodes := make([]*dht.Node, nodesToDiscv)
138 n := m.sw.GetDiscv().ReadRandomNodes(nodes)
139 for i := 0; i < n; i++ {
140 m.discvCh <- nodes[i]
145 func (m *monitor) collectDiscv() {
146 // nodeMap maps a node's public key to the node itself
147 nodeMap := make(map[string]*dht.Node)
148 for node := range m.discvCh {
149 if n, ok := nodeMap[node.ID.String()]; ok && n.String() == node.String() {
152 log.Info("discover new node: ", node)
154 m.upSertNode(&config.Node{
155 PublicKey: node.ID.String(),
156 Host: node.IP.String(),
159 nodeMap[node.ID.String()] = node
163 func (m *monitor) monitorRountine() error {
164 var nodes []*orm.Node
165 if err := m.db.Model(&orm.Node{}).Find(&nodes).Error; err != nil {
169 addresses := make([]*p2p.NetAddress, 0)
170 for i := 0; i < len(nodes); i++ {
171 ip, err := net.LookupIP(nodes[i].Host)
172 if err != nil || len(ip) == 0 {
176 address := p2p.NewNetAddressIPPort(ip[0], nodes[i].Port)
177 addresses = append(addresses, address)
180 m.sw.DialPeers(addresses)
182 // TODO: dail nodes, get lantency & best_height
183 // TODO: decide check_height("best best_height" - "confirmations")
184 // TODO: get blockhash by check_height, get latency
185 // TODO: update lantency, active_time and status
190 // implement logic first, and then refactor
191 // /home/gavin/work/go/src/github.com/vapor/
193 // p2p/switch_test.go