OSDN Git Service

Add mempool new tx subscription support (#1578)
[bytom/bytom.git] / node / node.go
1 package node
2
3 import (
4         "context"
5         "errors"
6         "net"
7         "net/http"
8         "os"
9         "path/filepath"
10
11         "github.com/prometheus/prometheus/util/flock"
12         log "github.com/sirupsen/logrus"
13         cmn "github.com/tendermint/tmlibs/common"
14         dbm "github.com/tendermint/tmlibs/db"
15         browser "github.com/toqueteos/webbrowser"
16
17         "github.com/bytom/accesstoken"
18         "github.com/bytom/account"
19         "github.com/bytom/api"
20         "github.com/bytom/asset"
21         "github.com/bytom/blockchain/pseudohsm"
22         "github.com/bytom/blockchain/txfeed"
23         cfg "github.com/bytom/config"
24         "github.com/bytom/consensus"
25         "github.com/bytom/database/leveldb"
26         "github.com/bytom/env"
27         "github.com/bytom/event"
28         "github.com/bytom/mining/cpuminer"
29         "github.com/bytom/mining/miningpool"
30         "github.com/bytom/mining/tensority"
31         "github.com/bytom/net/websocket"
32         "github.com/bytom/netsync"
33         "github.com/bytom/p2p"
34         "github.com/bytom/protocol"
35         w "github.com/bytom/wallet"
36 )
37
38 const webHost = "http://127.0.0.1"
39
40 // Node represent bytom node
41 type Node struct {
42         cmn.BaseService
43
44         config          *cfg.Config
45         eventDispatcher *event.Dispatcher
46         syncManager     *netsync.SyncManager
47
48         wallet          *w.Wallet
49         accessTokens    *accesstoken.CredentialStore
50         notificationMgr *websocket.WSNotificationManager
51         api             *api.API
52         chain           *protocol.Chain
53         txfeed          *txfeed.Tracker
54         cpuMiner        *cpuminer.CPUMiner
55         miningPool      *miningpool.MiningPool
56         miningEnable    bool
57 }
58
59 // NewNode create bytom node
60 func NewNode(config *cfg.Config) *Node {
61         ctx := context.Background()
62         if err := lockDataDirectory(config); err != nil {
63                 cmn.Exit("Error: " + err.Error())
64         }
65         initLogFile(config)
66         initActiveNetParams(config)
67         initCommonConfig(config)
68
69         // Get store
70         if config.DBBackend != "memdb" && config.DBBackend != "leveldb" {
71                 cmn.Exit(cmn.Fmt("Param db_backend [%v] is invalid, use leveldb or memdb", config.DBBackend))
72         }
73         coreDB := dbm.NewDB("core", config.DBBackend, config.DBDir())
74         store := leveldb.NewStore(coreDB)
75
76         tokenDB := dbm.NewDB("accesstoken", config.DBBackend, config.DBDir())
77         accessTokens := accesstoken.NewStore(tokenDB)
78
79         dispatcher := event.NewDispatcher()
80         txPool := protocol.NewTxPool(store, dispatcher)
81         chain, err := protocol.NewChain(store, txPool)
82         if err != nil {
83                 cmn.Exit(cmn.Fmt("Failed to create chain structure: %v", err))
84         }
85
86         var accounts *account.Manager
87         var assets *asset.Registry
88         var wallet *w.Wallet
89         var txFeed *txfeed.Tracker
90
91         txFeedDB := dbm.NewDB("txfeeds", config.DBBackend, config.DBDir())
92         txFeed = txfeed.NewTracker(txFeedDB, chain)
93
94         if err = txFeed.Prepare(ctx); err != nil {
95                 log.WithField("error", err).Error("start txfeed")
96                 return nil
97         }
98
99         hsm, err := pseudohsm.New(config.KeysDir())
100         if err != nil {
101                 cmn.Exit(cmn.Fmt("initialize HSM failed: %v", err))
102         }
103
104         if !config.Wallet.Disable {
105                 walletDB := dbm.NewDB("wallet", config.DBBackend, config.DBDir())
106                 accounts = account.NewManager(walletDB, chain)
107                 assets = asset.NewRegistry(walletDB, chain)
108                 wallet, err = w.NewWallet(walletDB, accounts, assets, hsm, chain, dispatcher)
109                 if err != nil {
110                         log.WithField("error", err).Error("init NewWallet")
111                 }
112
113                 // trigger rescan wallet
114                 if config.Wallet.Rescan {
115                         wallet.RescanBlocks()
116                 }
117         }
118
119         syncManager, err := netsync.NewSyncManager(config, chain, txPool, dispatcher)
120         if err != nil {
121                 cmn.Exit(cmn.Fmt("Failed to create sync manager: %v", err))
122         }
123
124         notificationMgr := websocket.NewWsNotificationManager(config.Websocket.MaxNumWebsockets, config.Websocket.MaxNumConcurrentReqs, chain, dispatcher)
125
126         // run the profile server
127         profileHost := config.ProfListenAddress
128         if profileHost != "" {
129                 // Profiling bytomd programs.see (https://blog.golang.org/profiling-go-programs)
130                 // go tool pprof http://profileHose/debug/pprof/heap
131                 go func() {
132                         if err = http.ListenAndServe(profileHost, nil); err != nil {
133                                 cmn.Exit(cmn.Fmt("Failed to register tcp profileHost: %v", err))
134                         }
135                 }()
136         }
137
138         node := &Node{
139                 eventDispatcher: dispatcher,
140                 config:          config,
141                 syncManager:     syncManager,
142                 accessTokens:    accessTokens,
143                 wallet:          wallet,
144                 chain:           chain,
145                 txfeed:          txFeed,
146                 miningEnable:    config.Mining.Enable,
147
148                 notificationMgr: notificationMgr,
149         }
150
151         node.cpuMiner = cpuminer.NewCPUMiner(chain, accounts, txPool, dispatcher)
152         node.miningPool = miningpool.NewMiningPool(chain, accounts, txPool, dispatcher, config.Mining.RecommitInterval)
153
154         node.BaseService = *cmn.NewBaseService(nil, "Node", node)
155
156         if config.Simd.Enable {
157                 tensority.UseSIMD = true
158         }
159
160         return node
161 }
162
163 // Lock data directory after daemonization
164 func lockDataDirectory(config *cfg.Config) error {
165         _, _, err := flock.New(filepath.Join(config.RootDir, "LOCK"))
166         if err != nil {
167                 return errors.New("datadir already used by another process")
168         }
169         return nil
170 }
171
172 func initActiveNetParams(config *cfg.Config) {
173         var exist bool
174         consensus.ActiveNetParams, exist = consensus.NetParams[config.ChainID]
175         if !exist {
176                 cmn.Exit(cmn.Fmt("chain_id[%v] don't exist", config.ChainID))
177         }
178 }
179
180 func initLogFile(config *cfg.Config) {
181         if config.LogFile == "" {
182                 return
183         }
184         cmn.EnsureDir(filepath.Dir(config.LogFile), 0700)
185         file, err := os.OpenFile(config.LogFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
186         if err == nil {
187                 log.SetOutput(file)
188         } else {
189                 log.WithField("err", err).Info("using default")
190         }
191
192 }
193
194 func initCommonConfig(config *cfg.Config) {
195         cfg.CommonConfig = config
196 }
197
198 // Lanch web broser or not
199 func launchWebBrowser(port string) {
200         webAddress := webHost + ":" + port
201         log.Info("Launching System Browser with :", webAddress)
202         if err := browser.Open(webAddress); err != nil {
203                 log.Error(err.Error())
204                 return
205         }
206 }
207
208 func (n *Node) initAndstartAPIServer() {
209         n.api = api.NewAPI(n.syncManager, n.wallet, n.txfeed, n.cpuMiner, n.miningPool, n.chain, n.config, n.accessTokens, n.eventDispatcher, n.notificationMgr)
210
211         listenAddr := env.String("LISTEN", n.config.ApiAddress)
212         env.Parse()
213         n.api.StartServer(*listenAddr)
214 }
215
216 func (n *Node) OnStart() error {
217         if n.miningEnable {
218                 if _, err := n.wallet.AccountMgr.GetMiningAddress(); err != nil {
219                         n.miningEnable = false
220                         log.Error(err)
221                 } else {
222                         n.cpuMiner.Start()
223                 }
224         }
225         if !n.config.VaultMode {
226                 if err := n.syncManager.Start(); err != nil {
227                         return err
228                 }
229         }
230
231         n.initAndstartAPIServer()
232         if err := n.notificationMgr.Start(); err != nil {
233                 return err
234         }
235
236         if !n.config.Web.Closed {
237                 _, port, err := net.SplitHostPort(n.config.ApiAddress)
238                 if err != nil {
239                         log.Error("Invalid api address")
240                         return err
241                 }
242                 launchWebBrowser(port)
243         }
244         return nil
245 }
246
247 func (n *Node) OnStop() {
248         n.notificationMgr.Shutdown()
249         n.notificationMgr.WaitForShutdown()
250         n.BaseService.OnStop()
251         if n.miningEnable {
252                 n.cpuMiner.Stop()
253         }
254         if !n.config.VaultMode {
255                 n.syncManager.Stop()
256         }
257         n.eventDispatcher.Stop()
258 }
259
260 func (n *Node) RunForever() {
261         // Sleep forever and then...
262         cmn.TrapSignal(func() {
263                 n.Stop()
264         })
265 }
266
267 func (n *Node) NodeInfo() *p2p.NodeInfo {
268         return n.syncManager.NodeInfo()
269 }
270
271 func (n *Node) MiningPool() *miningpool.MiningPool {
272         return n.miningPool
273 }