diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 522ad06b61..afa1d4d403 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -19,11 +19,13 @@ package utils import ( "crypto/ecdsa" + "errors" "fmt" "io/ioutil" "math/big" "os" "path/filepath" + "reflect" "runtime" "strconv" "strings" @@ -44,6 +46,7 @@ import ( "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/gasprice" + "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethstats" "github.com/ethereum/go-ethereum/les" @@ -1175,6 +1178,20 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) { } return fullNode, err }) + if err == nil { + err = stack.RegisterCallback(reflect.TypeOf(ð.Ethereum{}), func(service node.Service) error { + if e, ok := service.(*eth.Ethereum); ok { + rpcClient, err := stack.AttachLocked() + if err != nil { + return err + } + e.SetClient(ethclient.NewClient(rpcClient)) + return nil + } else { + return errors.New("the service given is not the required type") + } + }) + } } if err != nil { Fatalf("Failed to register the Ethereum service: %v", err) diff --git a/contracts/registrar/registrar.go b/contracts/registrar/registrar.go index 1c1a176b79..a7d8ce81c3 100644 --- a/contracts/registrar/registrar.go +++ b/contracts/registrar/registrar.go @@ -32,7 +32,7 @@ var ( RegistrarAddr = map[common.Hash]common.Address{ // params.MainnetGenesisHash: common.HexToAddress(""), // params.TestnetGenesisHash: common.HexToAddress(""), - params.RinkebyGenesisHash: common.HexToAddress("0xaefc742121f79ce9d0953fea7c9cd10f6586179b"), + params.RinkebyGenesisHash: common.HexToAddress("0xc72f57e41e2498ad3dab92f665b0f21e2c4f4b79"), } ) diff --git a/contracts/registrar/registrar_test.go b/contracts/registrar/registrar_test.go index 393ece8123..46f6ff1948 100644 --- a/contracts/registrar/registrar_test.go +++ b/contracts/registrar/registrar_test.go @@ -37,8 +37,8 @@ var ( addr = crypto.PubkeyToAddress(key.PublicKey) emptyHash = [32]byte{} - checkpointHash0 = crypto.Keccak256Hash(common.Hex2Bytes("dead0"), common.Hex2Bytes("beef0"), common.Hex2Bytes("deadbeef0")) - checkpointHash1 = crypto.Keccak256Hash(common.Hex2Bytes("dead1"), common.Hex2Bytes("beef1"), common.Hex2Bytes("deadbeef1")) + checkpointHash0 = crypto.Keccak256Hash(common.FromHex("dead0"), common.FromHex("beef0"), common.FromHex("deadbeef0")) + checkpointHash1 = crypto.Keccak256Hash(common.FromHex("dead1"), common.FromHex("beef1"), common.FromHex("deadbeef1")) ) // validateOperation executes the operation, watches and delivers all events fired by the backend and ensures the @@ -243,7 +243,7 @@ func TestCheckpointRegister(t *testing.T) { return nil }, "register stable checkpoint") - newHash := crypto.Keccak256Hash([]byte("dead00"), []byte("beef00"), []byte("deadbeef00")) + newHash := crypto.Keccak256Hash(common.FromHex("dead00"), common.FromHex("beef00"), common.FromHex("deadbeef00")) // Modify the latest checkpoint validateOperation(t, c, contractBackend, func() { c.SetCheckpoint(transactOpts, big.NewInt(0), newHash) @@ -280,7 +280,7 @@ func TestCheckpointRegister(t *testing.T) { }, "register stable checkpoint 1") contractBackend.ShiftBlocks(light.CheckpointConfirmations) - newHash1 := crypto.Keccak256Hash([]byte("dead11"), []byte("beef11"), []byte("deadbeef11")) + newHash1 := crypto.Keccak256Hash(common.FromHex("dead11"), common.FromHex("beef11"), common.FromHex("deadbeef11")) // Modify the registered checkpoint after a very long time validateOperation(t, c, contractBackend, func() { c.SetCheckpoint(transactOpts, big.NewInt(1), newHash1) diff --git a/eth/backend.go b/eth/backend.go index d601dd9c28..fabeee75fa 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -39,6 +39,7 @@ import ( "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/filters" "github.com/ethereum/go-ethereum/eth/gasprice" + "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/internal/ethapi" @@ -57,6 +58,7 @@ type LesServer interface { APIs() []rpc.API Protocols() []p2p.Protocol SetBloomBitsIndexer(bbIndexer *core.ChainIndexer) + SetClient(*ethclient.Client) } // Ethereum implements the Ethereum full node service. @@ -100,6 +102,14 @@ func (s *Ethereum) AddLesServer(ls LesServer) { ls.SetBloomBitsIndexer(s.bloomIndexer) } +// SetClient sets a rpc client which connecting to our local node. +func (s *Ethereum) SetClient(client *ethclient.Client) { + // Pass the rpc client to les server if it is enabled. + if s.lesServer != nil { + s.lesServer.SetClient(client) + } +} + // New creates a new Ethereum object (including the // initialisation of the common Ethereum object) func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { diff --git a/eth/bind.go b/eth/bind.go deleted file mode 100644 index ca339b696a..0000000000 --- a/eth/bind.go +++ /dev/null @@ -1,198 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package eth - -import ( - "context" - "math/big" - - "github.com/ethereum/go-ethereum" - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/hexutil" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/eth/filters" - "github.com/ethereum/go-ethereum/event" - "github.com/ethereum/go-ethereum/internal/ethapi" - "github.com/ethereum/go-ethereum/rlp" - "github.com/ethereum/go-ethereum/rpc" -) - -// ContractBackend implements bind.ContractBackend with direct calls to Ethereum -// internals to support operating on contracts within subprotocols like eth, les and -// swarm. -// -// Internally this backend uses the already exposed API endpoints of the Ethereum -// object. These should be rewritten to internal Go method calls when the Go API -// is refactored to support a clean library use. -type ContractBackend struct { - filterBackend filters.Backend // Backend used for filter vm logs. - events *filters.EventSystem // Event system used to watch new fired vm logs. - eapi *ethapi.PublicEthereumAPI // Wrapper around the Ethereum object to access metadata - bcapi *ethapi.PublicBlockChainAPI // Wrapper around the blockchain to access chain data - txapi *ethapi.PublicTransactionPoolAPI // Wrapper around the transaction pool to access transaction data -} - -// NewContractBackend creates a new native contract backend using an existing -// Ethereum object. -func NewContractBackend(apiBackend ethapi.Backend, filterBackend filters.Backend, lightMode bool) *ContractBackend { - return &ContractBackend{ - filterBackend: filterBackend, - events: filters.NewEventSystem(apiBackend.EventMux(), filterBackend, lightMode), - eapi: ethapi.NewPublicEthereumAPI(apiBackend), - bcapi: ethapi.NewPublicBlockChainAPI(apiBackend), - txapi: ethapi.NewPublicTransactionPoolAPI(apiBackend, new(ethapi.AddrLocker)), - } -} - -// CodeAt implements bind.ContractCaller retrieving any code associated -// with the contract from the local API. -func (b *ContractBackend) CodeAt(ctx context.Context, contract common.Address, blockNum *big.Int) ([]byte, error) { - return b.bcapi.GetCode(ctx, contract, toBlockNumber(blockNum)) -} - -// ContractCall implements bind.ContractCaller executing an Ethereum contract -// call with the specified data as the input. The pending flag requests execution -// against the pending block, not the stable head of the chain. -func (b *ContractBackend) CallContract(ctx context.Context, msg ethereum.CallMsg, blockNum *big.Int) ([]byte, error) { - out, err := b.bcapi.Call(ctx, toCallArgs(msg), toBlockNumber(blockNum)) - return out, err -} - -// PendingCodeAt implements bind.ContractTransactor retrieving any code associated -// with the contract from the local API. -func (b *ContractBackend) PendingCodeAt(ctx context.Context, contract common.Address) ([]byte, error) { - return b.bcapi.GetCode(ctx, contract, rpc.PendingBlockNumber) -} - -// PendingAccountNonce implements bind.ContractTransactor retrieving the current -// pending nonce associated with an account. -func (b *ContractBackend) PendingNonceAt(ctx context.Context, account common.Address) (nonce uint64, err error) { - out, err := b.txapi.GetTransactionCount(ctx, account, rpc.PendingBlockNumber) - if out != nil { - nonce = uint64(*out) - } - return nonce, err -} - -// SuggestGasPrice implements bind.ContractTransactor retrieving the currently -// suggested gas price to allow a timely execution of a transaction. -func (b *ContractBackend) SuggestGasPrice(ctx context.Context) (*big.Int, error) { - price, err := b.eapi.GasPrice(ctx) - if err != nil { - return nil, err - } - return price.ToInt(), nil -} - -// EstimateGasLimit implements bind.ContractTransactor trying to estimate the gas -// needed to execute a specific transaction based on the current pending state of -// the backend blockchain. There is no guarantee that this is the true gas limit -// requirement as other transactions may be added or removed by miners, but it -// should provide a basis for setting a reasonable default. -func (b *ContractBackend) EstimateGas(ctx context.Context, msg ethereum.CallMsg) (uint64, error) { - gas, err := b.bcapi.EstimateGas(ctx, toCallArgs(msg)) - return uint64(gas), err -} - -// SendTransaction implements bind.ContractTransactor injecting the transaction -// into the pending pool for execution. -func (b *ContractBackend) SendTransaction(ctx context.Context, tx *types.Transaction) error { - raw, _ := rlp.EncodeToBytes(tx) - _, err := b.txapi.SendRawTransaction(ctx, raw) - return err -} - -// FilterLogs implements bind.ContractFilterer returning logs matching the given argument -// that are stored within the state. -func (b *ContractBackend) FilterLogs(ctx context.Context, query ethereum.FilterQuery) ([]types.Log, error) { - // Initialize unset filter boundaried to run from genesis to chain head - from := int64(0) - if query.FromBlock != nil { - from = query.FromBlock.Int64() - } - to := int64(-1) - if query.ToBlock != nil { - to = query.ToBlock.Int64() - } - // Construct and execute the filter - filter := filters.New(b.filterBackend, from, to, query.Addresses, query.Topics) - - logs, err := filter.Logs(ctx) - if err != nil { - return nil, err - } - res := make([]types.Log, len(logs)) - for i, log := range logs { - res[i] = *log - } - return res, nil -} - -// SubscribeFilterLogs implements bind.ContractFilterer watching new fired logs matching the given argument. -func (b *ContractBackend) SubscribeFilterLogs(ctx context.Context, query ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error) { - // Subscribe to contract events - sink := make(chan []*types.Log) - sub, err := b.events.SubscribeLogs(query, sink) - if err != nil { - return nil, err - } - // Since we're getting logs in batches, we need to flatten them into a plain stream - return event.NewSubscription(func(quit <-chan struct{}) error { - defer sub.Unsubscribe() - for { - select { - case logs := <-sink: - for _, log := range logs { - select { - case ch <- *log: - case err := <-sub.Err(): - return err - case <-quit: - return nil - } - } - case err := <-sub.Err(): - return err - case <-quit: - return nil - } - } - }), nil -} - -func toCallArgs(msg ethereum.CallMsg) ethapi.CallArgs { - args := ethapi.CallArgs{ - To: msg.To, - From: msg.From, - Data: msg.Data, - Gas: hexutil.Uint64(msg.Gas), - } - if msg.GasPrice != nil { - args.GasPrice = hexutil.Big(*msg.GasPrice) - } - if msg.Value != nil { - args.Value = hexutil.Big(*msg.Value) - } - return args -} - -func toBlockNumber(num *big.Int) rpc.BlockNumber { - if num == nil { - return rpc.LatestBlockNumber - } - return rpc.BlockNumber(num.Int64()) -} diff --git a/les/server.go b/les/server.go index aeabbc0ff1..56bd04f61c 100644 --- a/les/server.go +++ b/les/server.go @@ -21,6 +21,8 @@ import ( "crypto/ecdsa" "errors" "sync" + "sync/atomic" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/contracts/registrar" @@ -29,6 +31,7 @@ import ( "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" + "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/les/flowcontrol" "github.com/ethereum/go-ethereum/light" @@ -53,9 +56,10 @@ type LesServer struct { privateKey *ecdsa.PrivateKey quitSync chan struct{} - // Checkpoint relative fields - registrar *registrar.Registrar // Handler for checkpoint contract - stableCheckpoint *light.TrustedCheckpoint // The nearest stable checkpoint + // Checkpoint contract relative fields + genesis common.Hash // Genesis block hash for contract address detection + registrar *registrar.Registrar // Handler for checkpoint contract, initialized after server is started. + watching int32 // Indicator whether the checkpoint contract is being watched // Indexers chtIndexer *core.ChainIndexer // Indexers for creating cht root for each block section @@ -83,6 +87,7 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) { lesTopics: lesTopics, chtIndexer: light.NewChtIndexer(e.ChainDb(), false), bloomTrieIndexer: light.NewBloomTrieIndexer(e.ChainDb(), false), + genesis: e.BlockChain().Genesis().Hash(), } logger := log.New() @@ -114,13 +119,6 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) { } srv.fcManager = flowcontrol.NewClientManager(uint64(config.LightServ), 10, 1000000000) srv.fcCostStats = newCostStats(e.ChainDb()) - if addr, ok := registrar.RegistrarAddr[e.BlockChain().Genesis().Hash()]; ok { - registrar, err := registrar.NewRegistrar(addr, eth.NewContractBackend(e.APIBackend, e.APIBackend, false)) - if err != nil { - return nil, err - } - srv.registrar = registrar - } return srv, nil } @@ -145,16 +143,28 @@ func (s *LesServer) Start(srvr *p2p.Server) { } s.privateKey = srvr.PrivateKey s.protocolManager.blockLoop() - if s.registrar != nil { - s.stableCheckpoint = s.recoverCheckpoint() - go s.checkpointLoop() - } } func (s *LesServer) SetBloomBitsIndexer(bloomIndexer *core.ChainIndexer) { bloomIndexer.AddChildIndexer(s.bloomTrieIndexer) } +// SetClient sets the rpc client and starts watching checkpoint contract if it is not yet watched. +func (s *LesServer) SetClient(client *ethclient.Client) { + if addr, ok := registrar.RegistrarAddr[s.genesis]; ok { + if !atomic.CompareAndSwapInt32(&s.watching, 0, 1) { + return + } + registrar, err := registrar.NewRegistrar(addr, client) + if err != nil { + atomic.StoreInt32(&s.watching, 0) + return + } + s.registrar = registrar + go s.checkpointLoop(s.recoverCheckpoint()) + } +} + // Stop stops the LES service func (s *LesServer) Stop() { s.chtIndexer.Close() @@ -165,6 +175,7 @@ func (s *LesServer) Stop() { <-s.protocolManager.noMorePeers }() s.protocolManager.Stop() + atomic.StoreInt32(&s.watching, 0) } // APIs implements LesServer, returns all API service provided by les server. @@ -215,7 +226,7 @@ func (s *LesServer) getCheckpoint(index uint64) (common.Hash, common.Hash, commo } // checkpointLoop starts a standalone goroutine to watch new checkpoint events and updates local's stable checkpoint. -func (s *LesServer) checkpointLoop() (err error) { +func (s *LesServer) checkpointLoop(checkpoint *light.TrustedCheckpoint) (err error) { var ( eventCh = make(chan *contract.ContractNewCheckpointEvent) headCh = make(chan core.ChainHeadEvent, SubscribeChainHeadEvent) @@ -230,9 +241,12 @@ func (s *LesServer) checkpointLoop() (err error) { eventSub.Unsubscribe() return errors.New("subscribe head event failed") } + + ticker := time.NewTicker(5 * time.Minute) defer func() { eventSub.Unsubscribe() headSub.Unsubscribe() + ticker.Stop() }() for { @@ -243,8 +257,9 @@ func (s *LesServer) checkpointLoop() (err error) { log.Info("Ignore empty checkpoint event") continue } - // Note several duplicate events may be received because of chain reorg and the modification of the latest checkpoint. - if s.stableCheckpoint == nil || event.Index.Uint64() >= s.stableCheckpoint.SectionIdx { + // Note several events have same index may be received because of chain reorg and + // the modification of the latest checkpoint. + if checkpoint == nil || event.Index.Uint64() > checkpoint.SectionIdx { log.Info("Receive new checkpoint event", "section", event.Index, "hash", common.Hash(event.CheckpointHash).Hex(), "grantor", event.Grantor.Hex()) announcement[event.Index.Uint64()] = common.Hash(event.CheckpointHash) @@ -255,22 +270,29 @@ func (s *LesServer) checkpointLoop() (err error) { continue } idx := (number-light.CheckpointConfirmations)/light.CheckpointFrequency - 1 - if s.stableCheckpoint == nil || idx > s.stableCheckpoint.SectionIdx { + if checkpoint == nil || idx > checkpoint.SectionIdx { hash, ok := announcement[idx] if !ok { continue } sectionHead := s.bloomTrieIndexer.SectionHead(idx) - checkpoint := &light.TrustedCheckpoint{ + c := &light.TrustedCheckpoint{ SectionIdx: idx, SectionHead: sectionHead, ChtRoot: light.GetChtV2Root(s.chaindb, idx, sectionHead), BloomTrieRoot: light.GetBloomTrieRoot(s.chaindb, idx, sectionHead), } - if checkpoint.HashEqual(common.Hash(hash)) { - light.WriteTrustedCheckpoint(s.chaindb, checkpoint) - s.stableCheckpoint = checkpoint - log.Info("Update stable checkpoint", "section", checkpoint.SectionIdx) + if c.HashEqual(common.Hash(hash)) { + light.WriteTrustedCheckpoint(s.chaindb, c) + checkpoint = c + delete(announcement, idx) + log.Info("Update stable checkpoint", "section", checkpoint.SectionIdx, "hash", checkpoint.Hash().Hex()) + } + } + case <-ticker.C: + // Evict useless announcement every 5 minutes. + for idx := range announcement { + if checkpoint != nil && checkpoint.SectionIdx >= idx { delete(announcement, idx) } } @@ -286,13 +308,13 @@ func (s *LesServer) recoverCheckpoint() *light.TrustedCheckpoint { var ( sectionCnt, _, _ = s.bloomTrieIndexer.Sections() stable = light.ReadTrustedCheckpoint(s.chaindb) - unstableIdx = sectionCnt - 1 headHash = rawdb.ReadHeadHeaderHash(s.chaindb) headNumber = rawdb.ReadHeaderNumber(s.chaindb, headHash) ) - if headNumber == nil { + if headNumber == nil || sectionCnt == 0 { return nil } + unstableIdx := sectionCnt - 1 for stable == nil || stable.SectionIdx < unstableIdx { if (unstableIdx+1)*light.CheckpointFrequency+light.CheckpointConfirmations <= *headNumber { iter, err := s.registrar.FilterNewCheckpointEvent(*headNumber, unstableIdx, light.CheckpointFrequency, light.CheckpointProcessConfirmations) @@ -308,7 +330,7 @@ func (s *LesServer) recoverCheckpoint() *light.TrustedCheckpoint { if checkpoint.HashEqual(common.Hash(iter.Event.CheckpointHash)) { light.WriteTrustedCheckpoint(s.chaindb, checkpoint) iter.Close() - log.Info("Recover checkpoint", "index", checkpoint.SectionIdx) + log.Info("Recover checkpoint", "index", checkpoint.SectionIdx, "hash", checkpoint.Hash().Hex()) return checkpoint } } @@ -323,7 +345,7 @@ func (s *LesServer) recoverCheckpoint() *light.TrustedCheckpoint { if stable == nil { log.Info("No stable checkpoint") } else { - log.Info("Recover checkpoint", "index", stable.SectionIdx) + log.Info("Recover checkpoint", "index", stable.SectionIdx, "hash", stable.Hash().Hex()) } return stable } diff --git a/light/checkpoint.go b/light/checkpoint.go index 118cb646b8..8933c84ac4 100644 --- a/light/checkpoint.go +++ b/light/checkpoint.go @@ -84,7 +84,12 @@ func (c *TrustedCheckpoint) HashEqual(hash common.Hash) bool { if c.SectionHead == (common.Hash{}) && c.ChtRoot == (common.Hash{}) && c.BloomTrieRoot == (common.Hash{}) { return hash == common.Hash{} } - return crypto.Keccak256Hash(c.SectionHead.Bytes(), c.ChtRoot.Bytes(), c.BloomTrieRoot.Bytes()) == hash + return c.Hash() == hash +} + +// Hash returns the hash of checkpoint three key fields(sectionHead, chtRoot and bloomTrieRoot). +func (c *TrustedCheckpoint) Hash() common.Hash { + return crypto.Keccak256Hash(c.SectionHead.Bytes(), c.ChtRoot.Bytes(), c.BloomTrieRoot.Bytes()) } var ( diff --git a/node/node.go b/node/node.go index ada3837217..58f9141d2a 100644 --- a/node/node.go +++ b/node/node.go @@ -48,8 +48,9 @@ type Node struct { serverConfig p2p.Config server *p2p.Server // Currently running P2P networking layer - serviceFuncs []ServiceConstructor // Service constructors (in dependency order) - services map[reflect.Type]Service // Currently running services + serviceFuncs []ServiceConstructor // Service constructors (in dependency order) + callbacks map[reflect.Type][]ServiceCallback // Service callback functions + services map[reflect.Type]Service // Currently running services rpcAPIs []rpc.API // List of APIs currently provided by the node inprocHandler *rpc.Server // In-process RPC request handler to process the API requests @@ -113,6 +114,7 @@ func New(conf *Config) (*Node, error) { ephemeralKeystore: ephemeralKeystore, config: conf, serviceFuncs: []ServiceConstructor{}, + callbacks: make(map[reflect.Type][]ServiceCallback), ipcEndpoint: conf.IPCEndpoint(), httpEndpoint: conf.HTTPEndpoint(), wsEndpoint: conf.WSEndpoint(), @@ -134,6 +136,18 @@ func (n *Node) Register(constructor ServiceConstructor) error { return nil } +// RegisterCallback injects a callback function associated with the specified service. +func (n *Node) RegisterCallback(typ reflect.Type, callback ServiceCallback) error { + n.lock.Lock() + defer n.lock.Unlock() + + if n.server != nil { + return ErrNodeRunning + } + n.callbacks[typ] = append(n.callbacks[typ], callback) + return nil +} + // Start create a live P2P node and starts running it. func (n *Node) Start() error { n.lock.Lock() @@ -211,7 +225,7 @@ func (n *Node) Start() error { // Mark the service started for potential cleanup started = append(started, kind) } - // Lastly start the configured RPC interfaces + // Start the configured RPC interfaces if err := n.startRPC(services); err != nil { for _, service := range services { service.Stop() @@ -223,7 +237,21 @@ func (n *Node) Start() error { n.services = services n.server = running n.stop = make(chan struct{}) - + // Lastly invokes all registered services callbacks after server is started. + for typ, callbacks := range n.callbacks { + if service, ok := services[typ]; ok { + for _, callback := range callbacks { + if err := callback(service); err != nil { + for _, service := range services { + service.Stop() + } + running.Stop() + n.services, n.server, n.stop = nil, nil, nil + return err + } + } + } + } return nil } @@ -480,7 +508,12 @@ func (n *Node) Restart() error { func (n *Node) Attach() (*rpc.Client, error) { n.lock.RLock() defer n.lock.RUnlock() + return n.AttachLocked() +} +// AttachLocked creates an RPC client attached to an in-process API handler. +// Note, this function assumes the lock of node is held. +func (n *Node) AttachLocked() (*rpc.Client, error) { if n.server == nil { return nil, ErrNodeStopped } diff --git a/node/service.go b/node/service.go index 6a96d9b1e1..ae65e10d4f 100644 --- a/node/service.go +++ b/node/service.go @@ -71,6 +71,10 @@ func (ctx *ServiceContext) Service(service interface{}) error { // registered for service instantiation. type ServiceConstructor func(ctx *ServiceContext) (Service, error) +// ServiceCallback is the function signature of the callbacks needed to be invoked +// after associated service is started. +type ServiceCallback func(service Service) error + // Service is an individual protocol that can be registered into a node. // // Notes: