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