node/node.go

   1 package node
   2
   3 import (
   4         "context"
   5         "net/http"
   6         _ "net/http/pprof"
   7         "strings"
   8         "time"
   9
  10         log "github.com/sirupsen/logrus"
  11         "github.com/tendermint/go-crypto"
  12         "github.com/tendermint/go-wire"
  13         cmn "github.com/tendermint/tmlibs/common"
  14         dbm "github.com/tendermint/tmlibs/db"
  15
  16         "github.com/bytom/accesstoken"
  17         "github.com/bytom/account"
  18         "github.com/bytom/api"
  19         "github.com/bytom/asset"
  20         bc "github.com/bytom/blockchain"
  21         "github.com/bytom/blockchain/pseudohsm"
  22         "github.com/bytom/blockchain/txfeed"
  23         cfg "github.com/bytom/config"
  24         "github.com/bytom/crypto/ed25519/chainkd"
  25         "github.com/bytom/database/leveldb"
  26         "github.com/bytom/env"
  27         "github.com/bytom/p2p"
  28         "github.com/bytom/protocol"
  29         "github.com/bytom/types"
  30         "github.com/bytom/util/browser"
  31         "github.com/bytom/version"
  32         w "github.com/bytom/wallet"
  33 )
  34
  35 const (
  36         webAddress               = "http://127.0.0.1:9888"
  37         expireReservationsPeriod = time.Second
  38 )
  39
  40 type Node struct {
  41         cmn.BaseService
  42
  43         // config
  44         config *cfg.Config
  45
  46         // network
  47         privKey  crypto.PrivKeyEd25519 // local node's p2p key
  48         sw       *p2p.Switch           // p2p connections
  49         addrBook *p2p.AddrBook         // known peers
  50
  51         evsw         types.EventSwitch // pub/sub for services
  52         bcReactor    *bc.BlockchainReactor
  53         wallet       *w.Wallet
  54         accessTokens *accesstoken.CredentialStore
  55         api          *api.API
  56         chain        *protocol.Chain
  57 }
  58
  59 func NewNode(config *cfg.Config) *Node {
  60         ctx := context.Background()
  61
  62         // Get store
  63         txDB := dbm.NewDB("txdb", config.DBBackend, config.DBDir())
  64         store := leveldb.NewStore(txDB)
  65
  66         tokenDB := dbm.NewDB("accesstoken", config.DBBackend, config.DBDir())
  67         accessTokens := accesstoken.NewStore(tokenDB)
  68
  69         privKey := crypto.GenPrivKeyEd25519()
  70
  71         // Make event switch
  72         eventSwitch := types.NewEventSwitch()
  73         _, err := eventSwitch.Start()
  74         if err != nil {
  75                 cmn.Exit(cmn.Fmt("Failed to start switch: %v", err))
  76         }
  77
  78         trustHistoryDB := dbm.NewDB("trusthistory", config.DBBackend, config.DBDir())
  79
  80         sw := p2p.NewSwitch(config.P2P, trustHistoryDB)
  81
  82         genesisBlock := cfg.GenerateGenesisBlock()
  83
  84         txPool := protocol.NewTxPool()
  85         chain, err := protocol.NewChain(genesisBlock.Hash(), store, txPool)
  86         if err != nil {
  87                 cmn.Exit(cmn.Fmt("Failed to create chain structure: %v", err))
  88         }
  89
  90         if chain.BestBlockHash() == nil {
  91                 if err := chain.SaveBlock(genesisBlock); err != nil {
  92                         cmn.Exit(cmn.Fmt("Failed to save genesisBlock to store: %v", err))
  93                 }
  94                 if err := chain.ConnectBlock(genesisBlock); err != nil {
  95                         cmn.Exit(cmn.Fmt("Failed to connect genesisBlock to chain: %v", err))
  96                 }
  97         }
  98
  99         var accounts *account.Manager = nil
 100         var assets *asset.Registry = nil
 101         var wallet *w.Wallet = nil
 102         var txFeed *txfeed.Tracker = nil
 103
 104         txFeedDB := dbm.NewDB("txfeeds", config.DBBackend, config.DBDir())
 105         txFeed = txfeed.NewTracker(txFeedDB, chain)
 106
 107         if err = txFeed.Prepare(ctx); err != nil {
 108                 log.WithField("error", err).Error("start txfeed")
 109                 return nil
 110         }
 111
 112         hsm, err := pseudohsm.New(config.KeysDir())
 113         if err != nil {
 114                 cmn.Exit(cmn.Fmt("initialize HSM failed: %v", err))
 115         }
 116
 117         if !config.Wallet.Disable {
 118                 walletDB := dbm.NewDB("wallet", config.DBBackend, config.DBDir())
 119                 accounts = account.NewManager(walletDB, chain)
 120                 assets = asset.NewRegistry(walletDB, chain)
 121                 wallet, err = w.NewWallet(walletDB, accounts, assets, hsm, chain)
 122                 if err != nil {
 123                         log.WithField("error", err).Error("init NewWallet")
 124                 }
 125
 126                 if err := initOrRecoverAccount(hsm, wallet); err != nil {
 127                         log.WithField("error", err).Error("initialize or recover account")
 128                 }
 129
 130                 // Clean up expired UTXO reservations periodically.
 131                 go accounts.ExpireReservations(ctx, expireReservationsPeriod)
 132         }
 133
 134         bcReactor := bc.NewBlockchainReactor(chain, txPool, sw, wallet, txFeed, config.Mining)
 135
 136         sw.AddReactor("BLOCKCHAIN", bcReactor)
 137
 138         // Optionally, start the pex reactor
 139         var addrBook *p2p.AddrBook
 140         if config.P2P.PexReactor {
 141                 addrBook = p2p.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
 142                 pexReactor := p2p.NewPEXReactor(addrBook)
 143                 sw.AddReactor("PEX", pexReactor)
 144         }
 145
 146         // run the profile server
 147         profileHost := config.ProfListenAddress
 148         if profileHost != "" {
 149                 // Profiling bytomd programs.see (https://blog.golang.org/profiling-go-programs)
 150                 // go tool pprof http://profileHose/debug/pprof/heap
 151                 go func() {
 152                         http.ListenAndServe(profileHost, nil)
 153                 }()
 154         }
 155
 156         node := &Node{
 157                 config: config,
 158
 159                 privKey:  privKey,
 160                 sw:       sw,
 161                 addrBook: addrBook,
 162
 163                 evsw:         eventSwitch,
 164                 bcReactor:    bcReactor,
 165                 accessTokens: accessTokens,
 166                 wallet:       wallet,
 167                 chain:        chain,
 168         }
 169         node.BaseService = *cmn.NewBaseService(nil, "Node", node)
 170
 171         return node
 172 }
 173
 174 func initOrRecoverAccount(hsm *pseudohsm.HSM, wallet *w.Wallet) error {
 175         xpubs := hsm.ListKeys()
 176
 177         if len(xpubs) == 0 {
 178                 xpub, err := hsm.XCreate("default", "123456")
 179                 if err != nil {
 180                         return err
 181                 }
 182
 183                 wallet.AccountMgr.Create(nil, []chainkd.XPub{xpub.XPub}, 1, "default", nil)
 184                 return nil
 185         }
 186
 187         accounts, err := wallet.AccountMgr.ListAccounts("")
 188         if err != nil {
 189                 return err
 190         }
 191
 192         if len(accounts) > 0 {
 193                 return nil
 194         }
 195
 196         for i, xPub := range xpubs {
 197                 if err := wallet.ImportAccountXpubKey(i, xPub, w.RecoveryIndex); err != nil {
 198                         return err
 199                 }
 200         }
 201         return nil
 202 }
 203
 204 // Lanch web broser or not
 205 func lanchWebBroser() {
 206         log.Info("Launching System Browser with :", webAddress)
 207         if err := browser.Open(webAddress); err != nil {
 208                 log.Error(err.Error())
 209                 return
 210         }
 211 }
 212
 213 func (n *Node) initAndstartApiServer() {
 214         n.api = api.NewAPI(n.bcReactor, n.wallet, n.chain, n.config, n.accessTokens)
 215
 216         listenAddr := env.String("LISTEN", n.config.ApiAddress)
 217         n.api.StartServer(*listenAddr)
 218 }
 219
 220 func (n *Node) OnStart() error {
 221         // Create & add listener
 222         p, address := ProtocolAndAddress(n.config.P2P.ListenAddress)
 223         l := p2p.NewDefaultListener(p, address, n.config.P2P.SkipUPNP, nil)
 224         n.sw.AddListener(l)
 225
 226         // Start the switch
 227         n.sw.SetNodeInfo(n.makeNodeInfo())
 228         n.sw.SetNodePrivKey(n.privKey)
 229         _, err := n.sw.Start()
 230         if err != nil {
 231                 return err
 232         }
 233
 234         // If seeds exist, add them to the address book and dial out
 235         if n.config.P2P.Seeds != "" {
 236                 // dial out
 237                 seeds := strings.Split(n.config.P2P.Seeds, ",")
 238                 if err := n.DialSeeds(seeds); err != nil {
 239                         return err
 240                 }
 241         }
 242
 243         n.initAndstartApiServer()
 244         if !n.config.Web.Closed {
 245                 lanchWebBroser()
 246         }
 247
 248         return nil
 249 }
 250
 251 func (n *Node) OnStop() {
 252         n.BaseService.OnStop()
 253
 254         log.Info("Stopping Node")
 255         // TODO: gracefully disconnect from peers.
 256         n.sw.Stop()
 257
 258 }
 259
 260 func (n *Node) RunForever() {
 261         // Sleep forever and then...
 262         cmn.TrapSignal(func() {
 263                 n.Stop()
 264         })
 265 }
 266
 267 // Add a Listener to accept inbound peer connections.
 268 // Add listeners before starting the Node.
 269 // The first listener is the primary listener (in NodeInfo)
 270 func (n *Node) AddListener(l p2p.Listener) {
 271         n.sw.AddListener(l)
 272 }
 273
 274 func (n *Node) Switch() *p2p.Switch {
 275         return n.sw
 276 }
 277
 278 func (n *Node) EventSwitch() types.EventSwitch {
 279         return n.evsw
 280 }
 281
 282 func (n *Node) makeNodeInfo() *p2p.NodeInfo {
 283         nodeInfo := &p2p.NodeInfo{
 284                 PubKey:  n.privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
 285                 Moniker: n.config.Moniker,
 286                 Network: "bytom",
 287                 Version: version.Version,
 288                 Other: []string{
 289                         cmn.Fmt("wire_version=%v", wire.Version),
 290                         cmn.Fmt("p2p_version=%v", p2p.Version),
 291                 },
 292         }
 293
 294         if !n.sw.IsListening() {
 295                 return nodeInfo
 296         }
 297
 298         p2pListener := n.sw.Listeners()[0]
 299         p2pHost := p2pListener.ExternalAddress().IP.String()
 300         p2pPort := p2pListener.ExternalAddress().Port
 301         //rpcListenAddr := n.config.RPC.ListenAddress
 302
 303         // We assume that the rpcListener has the same ExternalAddress.
 304         // This is probably true because both P2P and RPC listeners use UPnP,
 305         // except of course if the rpc is only bound to localhost
 306         nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
 307         //nodeInfo.Other = append(nodeInfo.Other, cmn.Fmt("rpc_addr=%v", rpcListenAddr))
 308         return nodeInfo
 309 }
 310
 311 //------------------------------------------------------------------------------
 312
 313 func (n *Node) NodeInfo() *p2p.NodeInfo {
 314         return n.sw.NodeInfo()
 315 }
 316
 317 func (n *Node) DialSeeds(seeds []string) error {
 318         return n.sw.DialSeeds(n.addrBook, seeds)
 319 }
 320
 321 // Defaults to tcp
 322 func ProtocolAndAddress(listenAddr string) (string, string) {
 323         p, address := "tcp", listenAddr
 324         parts := strings.SplitN(address, "://", 2)
 325         if len(parts) == 2 {
 326                 p, address = parts[0], parts[1]
 327         }
 328         return p, address
 329 }
 330
 331 //------------------------------------------------------------------------------