From 417d6acb6e1f99342271b489abc104ea9e8e92ed Mon Sep 17 00:00:00 2001 From: Bas van Kervel Date: Wed, 4 Nov 2015 15:30:47 +0100 Subject: [PATCH] rpc: migrated official API --- accounts/account_manager.go | 15 + accounts/account_rpc.go | 70 ++++ cmd/utils/flags.go | 49 ++- common/types.go | 40 ++- core/blockchain_rpc.go | 329 +++++++++++++++++++ core/transaction_pool_rpc.go | 346 ++++++++++++++++++++ core/types/block.go | 4 + core/types/bloom9.go | 4 + core/vm/log.go | 17 + eth/downloader/downloader_rpc.go | 42 +++ eth/eth_rpc.go | 76 +++++ eth/filters/filter_rpc.go | 530 +++++++++++++++++++++++++++++++ eth/web3_rpc.go | 22 ++ ethdb/interface.go | 1 - miner/miner_rpc.go | 52 +++ miner/worker.go | 1 + p2p/p2p_rpc.go | 29 ++ rpc/api/mergedapi.go | 8 +- rpc/api/utils.go | 2 + rpc/comms/ipc.go | 18 +- rpc/v2/json.go | 76 +++-- rpc/v2/server.go | 91 ++++-- rpc/v2/server_test.go | 4 - rpc/v2/types.go | 210 +++++++++++- rpc/v2/types_test.go | 57 ++++ rpc/v2/utils.go | 45 ++- 26 files changed, 2065 insertions(+), 73 deletions(-) create mode 100644 accounts/account_rpc.go create mode 100644 core/blockchain_rpc.go create mode 100644 core/transaction_pool_rpc.go create mode 100644 eth/downloader/downloader_rpc.go create mode 100644 eth/eth_rpc.go create mode 100644 eth/filters/filter_rpc.go create mode 100644 eth/web3_rpc.go create mode 100644 miner/miner_rpc.go create mode 100644 p2p/p2p_rpc.go create mode 100644 rpc/v2/types_test.go diff --git a/accounts/account_manager.go b/accounts/account_manager.go index 2781be656f..74006395cd 100644 --- a/accounts/account_manager.go +++ b/accounts/account_manager.go @@ -44,6 +44,10 @@ type Account struct { Address common.Address } +func (acc *Account) MarshalJSON() ([]byte, error) { + return []byte(`"` + acc.Address.Hex() + `"`), nil +} + type Manager struct { keyStore crypto.KeyStore unlocked map[common.Address]*unlocked @@ -92,6 +96,17 @@ func (am *Manager) Unlock(addr common.Address, keyAuth string) error { return am.TimedUnlock(addr, keyAuth, 0) } +func (am *Manager) Lock(addr common.Address) error { + am.mutex.Lock() + if unl, found := am.unlocked[addr]; found { + am.mutex.Unlock() + am.expire(addr, unl, time.Duration(0) * time.Nanosecond) + } else { + am.mutex.Unlock() + } + return nil +} + // TimedUnlock unlocks the account with the given address. The account // stays unlocked for the duration of timeout. A timeout of 0 unlocks the account // until the program exits. diff --git a/accounts/account_rpc.go b/accounts/account_rpc.go new file mode 100644 index 0000000000..0cb1bc16d8 --- /dev/null +++ b/accounts/account_rpc.go @@ -0,0 +1,70 @@ +package accounts +import ( + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/logger" + "time" +) + +// AccountService represents a RPC service with support for account specific actions. +type AccountService struct { + am *Manager +} + +// NewAccountService creates a new Account RPC service instance. +func NewAccountService(am *Manager) *AccountService { + return &AccountService{am: am} +} + +// Accounts returns the collection of accounts this node manages +func (s *AccountService) Accounts() ([]Account, error) { + return s.am.Accounts() +} + +// PersonalService represents a RPC service with support for personal methods. +type PersonalService struct { + am *Manager +} + +// NewPersonalService creates a new RPC service with support for personal actions. +func NewPersonalService(am *Manager) *PersonalService { + return &PersonalService{am} +} + +// ListAccounts will return a list of addresses for accounts this node manages. +func (s *PersonalService) ListAccounts(password string) ([]common.Address, error) { + accounts, err := s.am.Accounts() + if err != nil { + return nil, err + } + + addresses := make([]common.Address, len(accounts)) + for i, acc := range accounts { + addresses[i] = acc.Address + } + return addresses, nil +} + +// NewAccount will create a new account and returns the address for the new account. +func (s *PersonalService) NewAccount(password string) (common.Address, error) { + acc, err := s.am.NewAccount(password) + if err == nil { + return acc.Address, nil + } + return common.Address{}, err +} + +// UnlockAccount will unlock the account associated with the given address with the given password for duration seconds. +// It returns an indication if the action was successful. +func (s *PersonalService) UnlockAccount(addr common.Address, password string, duration int) bool { + if err := s.am.TimedUnlock(addr, password, time.Duration(duration) * time.Second); err != nil { + glog.V(logger.Info).Infof("%v\n", err) + return false + } + return true +} + +// LockAccount will lock the account associated with the given address when it's unlocked. +func (s *PersonalService) LockAccount(addr common.Address) bool { + return s.am.Lock(addr) == nil +} \ No newline at end of file diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 3792dc1e0d..1c2dd3552f 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -37,11 +37,14 @@ import ( "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth" + "github.com/ethereum/go-ethereum/eth/filters" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/miner" + "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/nat" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rpc/api" @@ -49,7 +52,9 @@ import ( "github.com/ethereum/go-ethereum/rpc/comms" "github.com/ethereum/go-ethereum/rpc/shared" "github.com/ethereum/go-ethereum/rpc/useragent" + rpc "github.com/ethereum/go-ethereum/rpc/v2" "github.com/ethereum/go-ethereum/xeth" + "github.com/ethereum/go-ethereum/eth/downloader" ) func init() { @@ -605,15 +610,51 @@ func IpcSocketPath(ctx *cli.Context) (ipcpath string) { return } -func StartIPC(eth *eth.Ethereum, ctx *cli.Context) error { +func StartIPC(ethereum *eth.Ethereum, ctx *cli.Context) error { + server := rpc.NewServer() + + server.RegisterName("eth", accounts.NewAccountService(ethereum.AccountManager())) + server.RegisterName("eth", core.NewBlockChainService(ethereum.BlockChain(), ethereum.AccountManager())) + server.RegisterName("eth", core.NewTransactionPoolService(ethereum.TxPool(), ethereum.ChainDb(), ethereum.BlockChain(), ethereum.AccountManager())) + server.RegisterName("eth", miner.NewMinerService(ethereum.Miner())) + server.RegisterName("eth", eth.NewEthService(ethereum)) + server.RegisterName("eth", downloader.NewDownloaderService(ethereum.Downloader())) + server.RegisterName("eth", filters.NewFilterService(ethereum.ChainDb(), ethereum.EventMux())) + server.RegisterName("net", p2p.NewNetService(ethereum.Network(), ethereum.NetVersion())) + server.RegisterName("web3", eth.NewWeb3Service(ethereum)) + server.RegisterName("personal", accounts.NewPersonalService(ethereum.AccountManager())) + + + ipcDir := filepath.Join(ethereum.DataDir, "shared") + os.MkdirAll(ipcDir, 0700) + ipcEndpoint := filepath.Join(ipcDir, "ethereum.sock") + os.RemoveAll(ipcEndpoint) + l, err := net.ListenUnix("unix", &net.UnixAddr{Name: ipcEndpoint, Net: "unix"}) + if err != nil { + panic(err) + } + go func() { + glog.Infof("Unix socket for IPC service opened on %s\n", ipcEndpoint) + for { + conn, err := l.Accept() + if err != nil { + panic(err) + } + codec := rpc.NewJSONCodec(conn) + go server.ServeCodec(codec) + } + }() + + config := comms.IpcConfig{ Endpoint: IpcSocketPath(ctx), } initializer := func(conn net.Conn) (comms.Stopper, shared.EthereumApi, error) { - fe := useragent.NewRemoteFrontend(conn, eth.AccountManager()) - xeth := xeth.New(eth, fe) - apis, err := api.ParseApiString(ctx.GlobalString(IPCApiFlag.Name), codec.JSON, xeth, eth) + fe := useragent.NewRemoteFrontend(conn, ethereum.AccountManager()) + xeth := xeth.New(ethereum, fe) + apis, err := api.ParseApiString(ctx.GlobalString(IPCApiFlag.Name), codec.JSON, xeth, ethereum) + if err != nil { return nil, nil, err } diff --git a/common/types.go b/common/types.go index e03d99830a..1ab9483410 100644 --- a/common/types.go +++ b/common/types.go @@ -17,11 +17,12 @@ package common import ( + "encoding/hex" + "encoding/json" "fmt" "math/big" "math/rand" "reflect" - "encoding/json" ) const ( @@ -51,6 +52,16 @@ func (h Hash) Bytes() []byte { return h[:] } func (h Hash) Big() *big.Int { return Bytes2Big(h[:]) } func (h Hash) Hex() string { return "0x" + Bytes2Hex(h[:]) } +// UnmarshalJSON parses a hash in its hex from to a hash. +func (h *Hash) UnmarshalJSON(input []byte) error { + length := len(input) + if length >= 2 && input[0] == '"' && input[length - 1] == '"' { + input = input[1:length - 1] + } + h.SetBytes(FromHex(string(input))) + return nil +} + // Serialize given hash to JSON func (h Hash) MarshalJSON() ([]byte, error) { return json.Marshal(h.Hex()) @@ -128,6 +139,33 @@ func (a Address) MarshalJSON() ([]byte, error) { return json.Marshal(a.Hex()) } +// Parse address from raw json data +func (a *Address) UnmarshalJSON(data []byte) error { + if len(data) > 2 && data[0] == '"' && data[len(data) - 1] == '"' { + data = data[:len(data) - 1][1:] + } + + if len(data) > 2 && data[0] == '0' && data[1] == 'x' { + data = data[2:] + } + + if len(data) != 2 * addressLength { + return fmt.Errorf("Invalid address length, expected %d got %d bytes", 2 * addressLength, len(data)) + } + + n, err := hex.Decode(a[:], data) + if err != nil { + return err + } + + if n != addressLength { + return fmt.Errorf("Invalid address") + } + + a.Set(HexToAddress(string(data))) + return nil +} + // PP Pretty Prints a byte slice in the following format: // hex(value[:4])...(hex[len(value)-4:]) func PP(value []byte) string { diff --git a/core/blockchain_rpc.go b/core/blockchain_rpc.go new file mode 100644 index 0000000000..7437bd6d1e --- /dev/null +++ b/core/blockchain_rpc.go @@ -0,0 +1,329 @@ +package core + +import ( + "fmt" + "math/big" + + "time" + + "github.com/ethereum/go-ethereum/accounts" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + rpc "github.com/ethereum/go-ethereum/rpc/v2" +) + +const ( + FilterTimeout = 300 * time.Second // Remove filter after FilterTimeout +) + +type BlockChainService struct { + bc *BlockChain + am *accounts.Manager +} + +func NewBlockChainService(bc *BlockChain, am *accounts.Manager) *BlockChainService { + return &BlockChainService{bc: bc, am: am} +} + +// BlockNumber returns the block number of the chain head. +func (s *BlockChainService) BlockNumber() *big.Int { + return s.bc.CurrentHeader().Number +} + +// GetBalance returns the amount of wei for the given address in the state of the given block number. +// When block number equals rpc.LatestBlockNumber the current block is used. +func (s *BlockChainService) GetBalance(address common.Address, blockNr rpc.BlockNumber) (*big.Int, error) { + block := blockByNumber(s.bc, blockNr) + if block == nil { + return nil, nil + } + + state, err := state.New(block.Root(), s.bc.chainDb) + if err != nil { + return nil, err + } + return state.GetBalance(address), nil +} + +// blockByNumber is a commonly used helper function which retrieves and returns the block for the given block number. It +// returns nil when no block could be found. +func blockByNumber(bc *BlockChain, blockNr rpc.BlockNumber) *types.Block { + if blockNr == rpc.LatestBlockNumber { + return bc.CurrentBlock() + } + + return bc.GetBlockByNumber(uint64(blockNr)) +} + +// GetBlockByNumber returns the requested block. When blockNr is -1 the chain head is returned. When fullTx is true all +// transactions in the block are returned in full detail, otherwise only the transaction hash is returned. +func (s *BlockChainService) GetBlockByNumber(blockNr rpc.BlockNumber, fullTx bool) (map[string]interface{}, error) { + if block := blockByNumber(s.bc, blockNr); block != nil { + return s.rpcOutputBlock(block, true, fullTx) + } + return nil, nil +} + +// GetBlockByHash returns the requested block. When fullTx is true all transactions in the block are returned in full +// detail, otherwise only the transaction hash is returned. +func (s *BlockChainService) GetBlockByHash(blockHash common.Hash, fullTx bool) (map[string]interface{}, error) { + if block := s.bc.GetBlock(blockHash); block != nil { + return s.rpcOutputBlock(block, true, fullTx) + } + return nil, nil +} + +// GetUncleByBlockNumberAndIndex returns the uncle block for the given block hash and index. When fullTx is true +// all transactions in the block are returned in full detail, otherwise only the transaction hash is returned. +func (s *BlockChainService) GetUncleByBlockNumberAndIndex(blockNr rpc.BlockNumber, index rpc.HexNumber) (map[string]interface{}, error) { + if blockNr == rpc.PendingBlockNumber { + return nil, nil + } + + if block := blockByNumber(s.bc, blockNr); block != nil { + uncles := block.Uncles() + if index.Int() < 0 || index.Int() >= len(uncles) { + glog.V(logger.Debug).Infof("uncle block on index %d not found for block #%d", index.Int(), blockNr) + return nil, nil + } + block = types.NewBlockWithHeader(uncles[index.Int()]) + return s.rpcOutputBlock(block, false, false) + } + return nil, nil +} + +// GetUncleByBlockHashAndIndex returns the uncle block for the given block hash and index. When fullTx is true +// all transactions in the block are returned in full detail, otherwise only the transaction hash is returned. +func (s *BlockChainService) GetUncleByBlockHashAndIndex(blockHash common.Hash, index rpc.HexNumber) (map[string]interface{}, error) { + if block := s.bc.GetBlock(blockHash); block != nil { + uncles := block.Uncles() + if index.Int() < 0 || index.Int() >= len(uncles) { + glog.V(logger.Debug).Infof("uncle block on index %d not found for block %s", index.Int(), blockHash.Hex()) + return nil, nil + } + block = types.NewBlockWithHeader(uncles[index.Int()]) + return s.rpcOutputBlock(block, false, false) + } + return nil, nil +} + +// GetUncleCountByBlockNumber returns number of uncles in the block for the given block number +func (s *BlockChainService) GetUncleCountByBlockNumber(blockNr rpc.BlockNumber) *rpc.HexNumber { + if blockNr == rpc.PendingBlockNumber { + return rpc.NewHexNumber(0) + } + + if block := blockByNumber(s.bc, blockNr); block != nil { + return rpc.NewHexNumber(len(block.Uncles())) + } + return nil +} + +// GetUncleCountByBlockHash returns number of uncles in the block for the given block hash +func (s *BlockChainService) GetUncleCountByBlockHash(blockHash common.Hash) *rpc.HexNumber { + if block := s.bc.GetBlock(blockHash); block != nil { + return rpc.NewHexNumber(len(block.Uncles())) + } + return nil +} + +// NewBlocksArgs allows the user to specify if the returned block should include transactions and in which format. +type NewBlocksArgs struct { + IncludeTransactions bool `json:"includeTransactions"` + TransactionDetails bool `json:"transactionDetails"` +} + +// NewBlocks triggers a new block event each time a block is appended to the chain. It accepts an argument which allows +// the caller to specify whether the output should contain transactions and in what format. +func (s *BlockChainService) NewBlocks(args NewBlocksArgs) (rpc.Subscription, error) { + sub := s.bc.eventMux.Subscribe(ChainEvent{}) + + output := func(rawBlock interface{}) interface{} { + if event, ok := rawBlock.(ChainEvent); ok { + notification, err := s.rpcOutputBlock(event.Block, args.IncludeTransactions, args.TransactionDetails) + if err == nil { + return notification + } + } + return rawBlock + } + + return rpc.NewSubscriptionWithOutputFormat(sub, output), nil +} + +func (s *BlockChainService) GetCode(address common.Address, blockNr rpc.BlockNumber) (string, error) { + if block := blockByNumber(s.bc, blockNr); block != nil { + state, err := state.New(block.Root(), s.bc.chainDb) + if err != nil { + return "", err + } + res := state.GetCode(address) + if len(res) == 0 { // backwards compatibility + return "0x", nil + } + return common.ToHex(res), nil + } + + return "0x", nil +} + +// StorageAt returns the storage at a given address +func (s *BlockChainService) GetStorageAt(address common.Address, position string, blockNr rpc.BlockNumber) (string, error) { + if block := blockByNumber(s.bc, blockNr); block != nil { + state, err := state.New(block.Root(), s.bc.chainDb) + if err != nil { + return "", err + } + return state.GetState(address, common.HexToHash(position)).Hex(), nil + } + + return "0x", nil +} + +// callmsg is the message type used for call transations. +type callmsg struct { + from *state.StateObject + to *common.Address + gas, gasPrice *big.Int + value *big.Int + data []byte +} + +// accessor boilerplate to implement core.Message +func (m callmsg) From() (common.Address, error) { return m.from.Address(), nil } +func (m callmsg) Nonce() uint64 { return m.from.Nonce() } +func (m callmsg) To() *common.Address { return m.to } +func (m callmsg) GasPrice() *big.Int { return m.gasPrice } +func (m callmsg) Gas() *big.Int { return m.gas } +func (m callmsg) Value() *big.Int { return m.value } +func (m callmsg) Data() []byte { return m.data } + +type CallArgs struct { + From common.Address `json:"from"` + To common.Address `json:"to"` + Gas rpc.HexNumber `json:"gas"` + GasPrice rpc.HexNumber `json:"gasPrice"` + Value rpc.HexNumber `json:"value"` + Data string `json:"data"` +} + +func (s *BlockChainService) doCall(args CallArgs, blockNr rpc.BlockNumber) (string, *big.Int, error) { + if block := blockByNumber(s.bc, blockNr); block != nil { + stateDb, err := state.New(block.Root(), s.bc.chainDb) + if err != nil { + return "0x", nil, err + } + + stateDb = stateDb.Copy() + var from *state.StateObject + if args.From == (common.Address{}) { + accounts, err := s.am.Accounts() + if err != nil || len(accounts) == 0 { + from = stateDb.GetOrNewStateObject(common.Address{}) + } else { + from = stateDb.GetOrNewStateObject(accounts[0].Address) + } + } else { + from = stateDb.GetOrNewStateObject(args.From) + } + + from.SetBalance(common.MaxBig) + + msg := callmsg{ + from: from, + to: &args.To, + gas: args.Gas.BigInt(), + gasPrice: args.GasPrice.BigInt(), + value: args.Value.BigInt(), + data: common.FromHex(args.Data), + } + + if msg.gas.Cmp(common.Big0) == 0 { + msg.gas = big.NewInt(50000000) + } + + if msg.gasPrice.Cmp(common.Big0) == 0 { + msg.gasPrice = new(big.Int).Mul(big.NewInt(50), common.Shannon) + } + + header := s.bc.CurrentBlock().Header() + vmenv := NewEnv(stateDb, s.bc, msg, header) + gp := new(GasPool).AddGas(common.MaxBig) + res, gas, err := ApplyMessage(vmenv, msg, gp) + if len(res) == 0 { // backwards compatability + return "0x", gas, err + } + return common.ToHex(res), gas, err + } + + return "0x", common.Big0, nil +} + +func (s *BlockChainService) Call(args CallArgs, blockNr rpc.BlockNumber) (string, error) { + result, _, err := s.doCall(args, blockNr) + return result, err +} + +func (s *BlockChainService) EstimateGas(args CallArgs) (*rpc.HexNumber, error) { + _, gas, err := s.doCall(args, rpc.LatestBlockNumber) + return rpc.NewHexNumber(gas), err +} + +// rpcOutputBlock converts the given block to the RPC output which depends on fullTx. If inclTx is true transactions are +// returned. When fullTx is true the returned block contains full transaction details, otherwise it will only contain +// transaction hashes. +func (s *BlockChainService) rpcOutputBlock(b *types.Block, inclTx bool, fullTx bool) (map[string]interface{}, error) { + fields := map[string]interface{}{ + "number": rpc.NewHexNumber(b.Number()), + "hash": b.Hash(), + "parentHash": b.ParentHash(), + "nonce": b.Header().Nonce, + "sha3Uncles": b.UncleHash(), + "logsBloom": b.Bloom(), + "stateRoot": b.Root(), + "miner": b.Coinbase(), + "difficulty": rpc.NewHexNumber(b.Difficulty()), + "totalDifficulty": rpc.NewHexNumber(s.bc.GetTd(b.Hash())), + "extraData": fmt.Sprintf("0x%x", b.Extra()), + "size": rpc.NewHexNumber(b.Size().Int64()), + "gasLimit": rpc.NewHexNumber(b.GasLimit()), + "gasUsed": rpc.NewHexNumber(b.GasUsed()), + "timestamp": rpc.NewHexNumber(b.Time()), + "transactionsRoot": b.TxHash(), + "receiptRoot": b.ReceiptHash(), + } + + if inclTx { + formatTx := func(tx *types.Transaction) (interface{}, error) { + return tx.Hash(), nil + } + + if fullTx { + formatTx = func(tx *types.Transaction) (interface{}, error) { + return newRPCTransaction(b, tx.Hash()) + } + } + + txs := b.Transactions() + transactions := make([]interface{}, len(txs)) + var err error + for i, tx := range b.Transactions() { + if transactions[i], err = formatTx(tx); err != nil { + return nil, err + } + } + fields["transactions"] = transactions + } + + uncles := b.Uncles() + uncleHashes := make([]common.Hash, len(uncles)) + for i, uncle := range uncles { + uncleHashes[i] = uncle.Hash() + } + fields["uncles"] = uncleHashes + + return fields, nil +} \ No newline at end of file diff --git a/core/transaction_pool_rpc.go b/core/transaction_pool_rpc.go new file mode 100644 index 0000000000..8a951c0cae --- /dev/null +++ b/core/transaction_pool_rpc.go @@ -0,0 +1,346 @@ +package core + +import ( + "bytes" + "fmt" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/vm" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/rlp" + rpc "github.com/ethereum/go-ethereum/rpc/v2" + "sync" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/accounts" + "github.com/ethereum/go-ethereum/event" +) + +const ( + defaultGasPrice = uint64(10000000000000) + defaultGas = uint64(90000) +) + +// RPCTransaction represents a transaction that will serialize to the RPC representation of a transaction +type RPCTransaction struct { + BlockHash common.Hash `json:"blockHash"` + BlockNumber *rpc.HexNumber `json:"blockNumber"` + From common.Address `json:"from"` + Gas *rpc.HexNumber `json:"gas"` + GasPrice *rpc.HexNumber `json:"gasPrice"` + Hash common.Hash `json:"hash"` + Input string `json:"input"` + Nonce *rpc.HexNumber `json:"nonce"` + To *common.Address `json:"to"` + TransactionIndex *rpc.HexNumber `json:"transactionIndex"` + Value *rpc.HexNumber `json:"value"` +} + +// newRPCTransaction returns a transaction that will serialize to the RPC representation. +func newRPCTransactionFromBlockIndex(b *types.Block, txIndex int) (*RPCTransaction, error) { + if txIndex >= 0 && txIndex < len(b.Transactions()) { + tx := b.Transactions()[txIndex] + from, err := tx.From() + if err != nil { + return nil, err + } + + return &RPCTransaction{ + BlockHash: b.Hash(), + BlockNumber: rpc.NewHexNumber(b.Number()), + From: from, + Gas: rpc.NewHexNumber(tx.Gas()), + GasPrice: rpc.NewHexNumber(tx.GasPrice()), + Hash: tx.Hash(), + Input: fmt.Sprintf("0x%x", tx.Data()), + Nonce: rpc.NewHexNumber(tx.Nonce()), + To: tx.To(), + TransactionIndex: rpc.NewHexNumber(txIndex), + Value: rpc.NewHexNumber(tx.Value()), + }, nil + } + + return nil, nil +} + +// newRPCTransaction returns a transaction that will serialize to the RPC representation. +func newRPCTransaction(b *types.Block, txHash common.Hash) (*RPCTransaction, error) { + for idx, tx := range b.Transactions() { + if tx.Hash() == txHash { + return newRPCTransactionFromBlockIndex(b, idx) + } + } + + return nil, nil +} + +// TransactionPoolService exposes methods for the RPC interface +type TransactionPoolService struct { + eventMux *event.TypeMux + chainDb ethdb.Database + bc *BlockChain + am *accounts.Manager + txPool *TxPool + txMu sync.Mutex +} + +// NewTransactionPoolService creates a new RPC service with methods specific for the transaction pool. +func NewTransactionPoolService(txPool *TxPool, chainDb ethdb.Database, bc *BlockChain, am *accounts.Manager) *TransactionPoolService { + return &TransactionPoolService{ + eventMux: txPool.eventMux, + chainDb: chainDb, + bc: bc, + am: am, + txPool: txPool, + } +} + +func getTransaction(chainDb ethdb.Database, txPool *TxPool, txHash common.Hash) (*types.Transaction, error) { + txData, err := chainDb.Get(txHash.Bytes()) + if err != nil { + return nil, err + } + + tx := new(types.Transaction) + if len(txData) > 0 { + if err := rlp.DecodeBytes(txData, tx); err != nil { + return nil, err + } + } else { // pending transaction? + tx = txPool.GetTransaction(txHash) + } + + return tx, nil +} + +// GetBlockTransactionCountByNumber returns the number of transactions in the block with the given block number. +func (s *TransactionPoolService) GetBlockTransactionCountByNumber(blockNr rpc.BlockNumber) *rpc.HexNumber { + if blockNr == rpc.PendingBlockNumber { + return rpc.NewHexNumber(0) + } + + if block := blockByNumber(s.bc, blockNr); block != nil { + return rpc.NewHexNumber(len(block.Transactions())) + } + + return nil +} + +// GetBlockTransactionCountByHash returns the number of transactions in the block with the given hash. +func (s *TransactionPoolService) GetBlockTransactionCountByHash(blockHash common.Hash) *rpc.HexNumber { + if block := s.bc.GetBlock(blockHash); block != nil { + return rpc.NewHexNumber(len(block.Transactions())) + } + return nil +} + +// GetTransactionByBlockNumberAndIndex returns the transaction for the given block number and index. +func (s *TransactionPoolService) GetTransactionByBlockNumberAndIndex(blockNr rpc.BlockNumber, index rpc.HexNumber) (*RPCTransaction, error) { + if block := blockByNumber(s.bc, blockNr); block != nil { + return newRPCTransactionFromBlockIndex(block, index.Int()) + } + return nil, nil +} + +// GetTransactionByBlockHashAndIndex returns the transaction for the given block hash and index. +func (s *TransactionPoolService) GetTransactionByBlockHashAndIndex(blockHash common.Hash, index rpc.HexNumber) (*RPCTransaction, error) { + if block := s.bc.GetBlock(blockHash); block != nil { + return newRPCTransactionFromBlockIndex(block, index.Int()) + } + return nil, nil +} + +// GetTransactionCount returns the number of transactions the given address has sent for the given block number +func (s *TransactionPoolService) GetTransactionCount(address common.Address, blockNr rpc.BlockNumber) (*rpc.HexNumber, error) { + block := blockByNumber(s.bc, blockNr) + if block == nil { + return nil, nil + } + + state, err := state.New(block.Root(), s.chainDb) + if err != nil { + return nil, err + } + return rpc.NewHexNumber(state.GetNonce(address)), nil +} + +// getTransactionBlockData fetches the meta data for the given transaction from the chain database. This is useful to +// retrieve block information for a hash. It returns the block hash, block index and transaction index. +func getTransactionBlockData(chainDb ethdb.Database, txHash common.Hash) (common.Hash, uint64, uint64, error) { + var txBlock struct { + BlockHash common.Hash + BlockIndex uint64 + Index uint64 + } + + blockData, err := chainDb.Get(append(txHash.Bytes(), 0x0001)) + if err != nil { + return common.Hash{}, uint64(0), uint64(0), err + } + + reader := bytes.NewReader(blockData) + if err = rlp.Decode(reader, &txBlock); err != nil { + return common.Hash{}, uint64(0), uint64(0), err + } + + return txBlock.BlockHash, txBlock.BlockIndex, txBlock.Index, nil +} + +// GetTransactionByHash returns the transaction for the given hash +func (s *TransactionPoolService) GetTransactionByHash(txHash common.Hash) (*RPCTransaction, error) { + var tx *types.Transaction + var err error + + if tx, err = getTransaction(s.chainDb, s.txPool, txHash); err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + return nil, nil + } else if tx == nil { + return nil, nil + } + + blockHash, _, _, err := getTransactionBlockData(s.chainDb, txHash) + if err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + return nil, nil + } + + if block := s.bc.GetBlock(blockHash); block != nil { + return newRPCTransaction(block, txHash) + } + + return nil, nil +} + +// GetTransactionReceipt returns the transaction receipt for the given transaction hash. +func (s *TransactionPoolService) GetTransactionReceipt(txHash common.Hash) (map[string]interface{}, error) { + receipt := GetReceipt(s.chainDb, txHash) + if receipt == nil { + glog.V(logger.Debug).Infof("receipt not found for transaction %s", txHash.Hex()) + return nil, nil + } + + tx, err := getTransaction(s.chainDb, s.txPool, txHash) + if err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + return nil, nil + } + + txBlock, blockIndex, index, err := getTransactionBlockData(s.chainDb, txHash) + if err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + return nil, nil + } + + from, err := tx.From() + if err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + return nil, nil + } + + fields := map[string]interface{}{ + "blockHash": txBlock, + "blockNumber": rpc.NewHexNumber(blockIndex), + "transactionHash": txHash, + "transactionIndex": rpc.NewHexNumber(index), + "from": from, + "to": tx.To(), + "gasUsed": rpc.NewHexNumber(receipt.GasUsed), + "cumulativeGasUsed": rpc.NewHexNumber(receipt.CumulativeGasUsed), + "contractAddress": nil, + "logs": receipt.Logs, + } + + if receipt.Logs == nil { + fields["logs"] = []vm.Logs{} + } + + // If the ContractAddress is 20 0x0 bytes, assume it is not a contract creation + if bytes.Compare(receipt.ContractAddress.Bytes(), bytes.Repeat([]byte{0}, 20)) != 0 { + fields["contractAddress"] = receipt.ContractAddress + } + + return fields, nil +} + + +func (s *TransactionPoolService) sign(from common.Address, tx *types.Transaction) (*types.Transaction, error) { + acc := accounts.Account{from} + signature, err := s.am.Sign(acc, tx.SigHash().Bytes()) + if err != nil { + return nil, err + } + + return tx.WithSignature(signature) +} + +type SendTxArgs struct { + From common.Address `json:"from"` + To common.Address `json:"to"` + Gas *rpc.HexNumber `json:"gas"` + GasPrice *rpc.HexNumber `json:"gasPrice"` + Value *rpc.HexNumber `json:"value"` + Data string `json:"data"` + Nonce *rpc.HexNumber `json:"nonce"` +} + +func (s *TransactionPoolService) SendTransaction(args SendTxArgs) (common.Hash, error) { + if args.Gas == nil { + args.Gas = rpc.NewHexNumber(defaultGas) + } + if args.GasPrice == nil { + args.GasPrice = rpc.NewHexNumber(defaultGasPrice) + } + if args.Value == nil { + args.Value = rpc.NewHexNumber(0) + } + + s.txMu.Lock() + defer s.txMu.Unlock() + + if args.Nonce == nil { + args.Nonce = rpc.NewHexNumber(s.txPool.State().GetNonce(args.From)) + } + + var tx *types.Transaction + contractCreation := (args.To == common.Address{}) + + if contractCreation { + tx = types.NewContractCreation(args.Nonce.Uint64(), args.Value.BigInt(), args.Gas.BigInt(), args.GasPrice.BigInt(), common.FromHex(args.Data)) + } else { + tx = types.NewTransaction(args.Nonce.Uint64(), args.To, args.Value.BigInt(), args.Gas.BigInt(), args.GasPrice.BigInt(), common.FromHex(args.Data)) + } + + signedTx, err := s.sign(args.From, tx) + if err != nil { + return common.Hash{}, err + } + + if err := s.txPool.Add(signedTx); err != nil { + return common.Hash{}, nil + } + + if contractCreation { + addr := crypto.CreateAddress(args.From, args.Nonce.Uint64()) + glog.V(logger.Info).Infof("Tx(%s) created: %s\n", signedTx.Hash().Hex(), addr.Hex()) + } else { + glog.V(logger.Info).Infof("Tx(%s) to: %s\n", signedTx.Hash().Hex(), tx.To().Hex()) + } + + return signedTx.Hash(), nil +} + +func (s *TransactionPoolService) PendingTransactions() (rpc.Subscription, error) { + sub := s.eventMux.Subscribe(TxPreEvent{}) + + output := func(transaction interface{}) interface{} { + if tx, ok := transaction.(TxPreEvent); ok { + return tx.Tx.Hash() + } + return nil + } + + return rpc.NewSubscriptionWithOutputFormat(sub, output), nil +} diff --git a/core/types/block.go b/core/types/block.go index 1d1cfa5151..5536e0ea86 100644 --- a/core/types/block.go +++ b/core/types/block.go @@ -48,6 +48,10 @@ func (n BlockNonce) Uint64() uint64 { return binary.BigEndian.Uint64(n[:]) } +func (n BlockNonce) MarshalJSON() ([]byte, error) { + return []byte(fmt.Sprintf(`"0x%x"`, n)), nil +} + type Header struct { ParentHash common.Hash // Hash to the previous block UncleHash common.Hash // Uncles of this block diff --git a/core/types/bloom9.go b/core/types/bloom9.go index cd90fd9716..372045ab21 100644 --- a/core/types/bloom9.go +++ b/core/types/bloom9.go @@ -69,6 +69,10 @@ func (b Bloom) TestBytes(test []byte) bool { return b.Test(common.BytesToBig(test)) } +func (b Bloom) MarshalJSON() ([]byte, error) { + return []byte(fmt.Sprintf(`"%#x"`, b.Bytes())), nil +} + func CreateBloom(receipts Receipts) Bloom { bin := new(big.Int) for _, receipt := range receipts { diff --git a/core/vm/log.go b/core/vm/log.go index 191e3a2539..a8e73e17b8 100644 --- a/core/vm/log.go +++ b/core/vm/log.go @@ -17,11 +17,13 @@ package vm import ( + "encoding/json" "fmt" "io" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/rlp" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) type Log struct { @@ -63,6 +65,21 @@ func (l *Log) String() string { return fmt.Sprintf(`log: %x %x %x %x %d %x %d`, l.Address, l.Topics, l.Data, l.TxHash, l.TxIndex, l.BlockHash, l.Index) } +func (r *Log) MarshalJSON() ([]byte, error) { + fields := map[string]interface{}{ + "address": r.Address, + "data": fmt.Sprintf("%#x", r.Data), + "blockNumber": rpc.NewHexNumber(r.BlockNumber), + "logIndex": rpc.NewHexNumber(r.Index), + "blockHash": r.BlockHash, + "transactionHash": r.TxHash, + "transactionIndex": rpc.NewHexNumber(r.TxIndex), + "topics": r.Topics, + } + + return json.Marshal(fields) +} + type Logs []*Log // LogForStorage is a wrapper around a Log that flattens and parses the entire diff --git a/eth/downloader/downloader_rpc.go b/eth/downloader/downloader_rpc.go new file mode 100644 index 0000000000..7eb1ae77f5 --- /dev/null +++ b/eth/downloader/downloader_rpc.go @@ -0,0 +1,42 @@ +package downloader + +import ( + rpc "github.com/ethereum/go-ethereum/rpc/v2" +) + +type DownloaderService struct { + d *Downloader +} + +func NewDownloaderService(d *Downloader) *DownloaderService { + return &DownloaderService{d} +} + +type Progress struct { + Origin uint64 `json:"startingBlock"` + Current uint64 `json:"currentBlock"` + Height uint64 `json:"highestBlock"` +} + +type SyncingResult struct { + Syncing bool `json:"syncing"` + Status Progress `json:"status"` +} + +func (s *DownloaderService) Syncing() (rpc.Subscription, error) { + sub := s.d.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{}) + + output := func(event interface{}) interface{} { + switch event.(type) { + case StartEvent: + result := &SyncingResult{Syncing: true} + result.Status.Origin, result.Status.Current, result.Status.Height = s.d.Progress() + return result + case DoneEvent, FailedEvent: + return false + } + return nil + } + + return rpc.NewSubscriptionWithOutputFormat(sub, output), nil +} diff --git a/eth/eth_rpc.go b/eth/eth_rpc.go new file mode 100644 index 0000000000..1aae5177eb --- /dev/null +++ b/eth/eth_rpc.go @@ -0,0 +1,76 @@ +package eth + +import ( + "math/big" + "errors" + + "github.com/ethereum/go-ethereum/common/compiler" + "github.com/ethereum/go-ethereum/common" + rpc "github.com/ethereum/go-ethereum/rpc/v2" +) + +type EthService struct { + e *Ethereum + gpo *GasPriceOracle +} + +func NewEthService(e *Ethereum) *EthService { + return &EthService{e, NewGasPriceOracle(e)} +} + +// GasPrice returns a suggestion for a gas price. +func (s *EthService) GasPrice() *big.Int { + return s.gpo.SuggestPrice() +} + +// GetCompilers returns the collection of available smart contract compilers +func (s *EthService) GetCompilers() ([]string, error) { + solc, err := s.e.Solc() + if err != nil { + return nil, err + } + + if solc != nil { + return []string{"Solidity"}, nil + } + + return nil, nil +} + +// CompileSolidity compiles the given solidity source +func (s *EthService) CompileSolidity(source string) (map[string]*compiler.Contract, error) { + solc, err := s.e.Solc() + if err != nil { + return nil, err + } + + if solc == nil { + return nil, errors.New("solc (solidity compiler) not found") + } + + return solc.Compile(source) +} + +func (s *EthService) Etherbase() (common.Address, error) { + return s.e.Etherbase() +} + +func (s *EthService) Coinbase() (common.Address, error) { + return s.Etherbase() +} + +func (s *EthService) ProtocolVersion() *rpc.HexNumber { + return rpc.NewHexNumber(s.e.EthVersion()) +} + +func (s *EthService) Syncing() (interface{}, error) { + origin, current, height := s.e.Downloader().Progress() + if current < height { + return map[string]interface{}{ + "startingBlock": rpc.NewHexNumber(origin), + "currentBlock": rpc.NewHexNumber(current), + "highestBlock": rpc.NewHexNumber(height), + }, nil + } + return false, nil +} diff --git a/eth/filters/filter_rpc.go b/eth/filters/filter_rpc.go new file mode 100644 index 0000000000..141381c1b2 --- /dev/null +++ b/eth/filters/filter_rpc.go @@ -0,0 +1,530 @@ +package filters + +import ( + "sync" + "time" + + "crypto/rand" + "encoding/hex" + "errors" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/vm" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/core/types" + rpc "github.com/ethereum/go-ethereum/rpc/v2" + "encoding/json" + "fmt" +) + +var ( + filterTickerTime = 5 * time.Minute +) + +// byte will be inferred +const ( + unknownFilterTy = iota + blockFilterTy + transactionFilterTy + logFilterTy +) + + +type FilterService struct { + mux *event.TypeMux + + quit chan struct{} + chainDb ethdb.Database + + filterManager *FilterSystem + + filterMapMu sync.RWMutex + filterMapping map[string]int // maps between filter internal filter identifiers and external filter identifiers + + logMu sync.RWMutex + logQueue map[int]*logQueue + + blockMu sync.RWMutex + blockQueue map[int]*hashQueue + + transactionMu sync.RWMutex + transactionQueue map[int]*hashQueue + + // messagesMu sync.RWMutex + // messages map[int]*whisperFilter + + transactMu sync.Mutex +} + +func NewFilterService(chainDb ethdb.Database, mux *event.TypeMux) *FilterService { + svc := &FilterService{ + mux: mux, + chainDb: chainDb, + filterManager: NewFilterSystem(mux), + filterMapping: make(map[string]int), + logQueue: make(map[int]*logQueue), + blockQueue: make(map[int]*hashQueue), + transactionQueue: make(map[int]*hashQueue), + } + go svc.start() + return svc +} + +func (s *FilterService) Stop() { + close(s.quit) +} + +func (s *FilterService) start() { + timer := time.NewTicker(2 * time.Second) + defer timer.Stop() + done: + for { + select { + case <-timer.C: + s.logMu.Lock() + for id, filter := range s.logQueue { + if time.Since(filter.timeout) > filterTickerTime { + s.filterManager.Remove(id) + delete(s.logQueue, id) + } + } + s.logMu.Unlock() + + s.blockMu.Lock() + for id, filter := range s.blockQueue { + if time.Since(filter.timeout) > filterTickerTime { + s.filterManager.Remove(id) + delete(s.blockQueue, id) + } + } + s.blockMu.Unlock() + + s.transactionMu.Lock() + for id, filter := range s.transactionQueue { + if time.Since(filter.timeout) > filterTickerTime { + s.filterManager.Remove(id) + delete(s.transactionQueue, id) + } + } + s.transactionMu.Unlock() + + // s.messagesMu.Lock() + // for id, filter := range s.messages { + // if time.Since(filter.activity()) > filterTickerTime { + // s.Whisper().Unwatch(id) + // delete(s.messages, id) + // } + // } + // s.messagesMu.Unlock() + case <-s.quit: + break done + } + } + +} + +func (s *FilterService) NewBlockFilter() (string, error) { + externalId, err := newFilterId() + if err != nil { + return "", err + } + + s.blockMu.Lock() + filter := New(s.chainDb) + id := s.filterManager.Add(filter) + s.blockQueue[id] = &hashQueue{timeout: time.Now()} + + filter.BlockCallback = func(block *types.Block, logs vm.Logs) { + s.blockMu.Lock() + defer s.blockMu.Unlock() + + if queue := s.blockQueue[id]; queue != nil { + queue.add(block.Hash()) + } + } + + defer s.blockMu.Unlock() + + s.filterMapMu.Lock() + s.filterMapping[externalId] = id + s.filterMapMu.Unlock() + + return externalId, nil +} + +func (s *FilterService) NewPendingTransactionFilter() (string, error) { + externalId, err := newFilterId() + if err != nil { + return "", err + } + + s.transactionMu.Lock() + defer s.transactionMu.Unlock() + + filter := New(s.chainDb) + id := s.filterManager.Add(filter) + s.transactionQueue[id] = &hashQueue{timeout: time.Now()} + + filter.TransactionCallback = func(tx *types.Transaction) { + s.transactionMu.Lock() + defer s.transactionMu.Unlock() + + if queue := s.transactionQueue[id]; queue != nil { + queue.add(tx.Hash()) + } + } + + s.filterMapMu.Lock() + s.filterMapping[externalId] = id + s.filterMapMu.Unlock() + + return externalId, nil +} + +func (s *FilterService) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash) int { + s.logMu.Lock() + defer s.logMu.Unlock() + + filter := New(s.chainDb) + id := s.filterManager.Add(filter) + s.logQueue[id] = &logQueue{timeout: time.Now()} + + filter.SetBeginBlock(earliest) + filter.SetEndBlock(latest) + filter.SetAddresses(addresses) + filter.SetTopics(topics) + filter.LogsCallback = func(logs vm.Logs) { + s.logMu.Lock() + defer s.logMu.Unlock() + + if queue := s.logQueue[id]; queue != nil { + queue.add(logs...) + } + } + + return id +} + +type NewFilterArgs struct { + FromBlock rpc.BlockNumber + ToBlock rpc.BlockNumber + Addresses []common.Address + Topics [][]common.Hash +} + +func (args *NewFilterArgs) UnmarshalJSON(data []byte) error { + type input struct { + From *rpc.BlockNumber `json:"fromBlock"` + ToBlock *rpc.BlockNumber `json:"toBlock"` + Addresses interface{} `json:"address"` + Topics interface{} `json:"topics"` + } + + var raw input + if err := json.Unmarshal(data, &raw); err != nil { + return err + } + + if raw.From == nil { + args.FromBlock = rpc.LatestBlockNumber + } else { + args.FromBlock = *raw.From + } + + if raw.ToBlock == nil { + args.ToBlock = rpc.LatestBlockNumber + } else { + args.ToBlock = *raw.ToBlock + } + + args.Addresses = []common.Address{} + + if raw.Addresses != nil { + // raw.Address can contain a single address or an array of addresses + var addresses []common.Address + + if strAddrs, ok := raw.Addresses.([]interface{}); ok { + for i, addr := range strAddrs { + if strAddr, ok := addr.(string); ok { + if len(strAddr) >= 2 && strAddr[0] == '0' && (strAddr[1] == 'x' || strAddr[1] == 'X') { + strAddr = strAddr[2:] + } + if decAddr, err := hex.DecodeString(strAddr); err == nil { + addresses = append(addresses, common.BytesToAddress(decAddr)) + } else { + fmt.Errorf("invalid address given") + } + } else { + return fmt.Errorf("invalid address on index %d", i) + } + } + } else if singleAddr, ok := raw.Addresses.(string); ok { + if len(singleAddr) >= 2 && singleAddr[0] == '0' && (singleAddr[1] == 'x' || singleAddr[1] == 'X') { + singleAddr = singleAddr[2:] + } + if decAddr, err := hex.DecodeString(singleAddr); err == nil { + addresses = append(addresses, common.BytesToAddress(decAddr)) + } else { + fmt.Errorf("invalid address given") + } + } else { + errors.New("invalid address(es) given") + } + args.Addresses = addresses + } + + topicConverter := func(raw string) (common.Hash, error) { + if len(raw) == 0 { + return common.Hash{}, nil + } + + if len(raw) >= 2 && raw[0] == '0' && (raw[1] == 'x' || raw[1] == 'X') { + raw = raw[2:] + } + + if decAddr, err := hex.DecodeString(raw); err == nil { + return common.BytesToHash(decAddr), nil + } + + return common.Hash{}, errors.New("invalid topic given") + } + + // topics is an array consisting of strings or arrays of strings + if raw.Topics != nil { + topics, ok := raw.Topics.([]interface{}) + if ok { + parsedTopics := make([][]common.Hash, len(topics)) + for i, topic := range topics { + if topic == nil { + parsedTopics[i] = []common.Hash{common.StringToHash("")} + } else if strTopic, ok := topic.(string); ok { + if t, err := topicConverter(strTopic); err != nil { + return fmt.Errorf("invalid topic on index %d", i) + } else { + parsedTopics[i] = []common.Hash{t} + } + } else if arrTopic, ok := topic.([]interface{}); ok { + parsedTopics[i] = make([]common.Hash, len(arrTopic)) + for j := 0; j < len(parsedTopics[i]); i++ { + if arrTopic[j] == nil { + parsedTopics[i][j] = common.StringToHash("") + } else if str, ok := arrTopic[j].(string); ok { + if t, err := topicConverter(str); err != nil { + return fmt.Errorf("invalid topic on index %d", i) + } else { + parsedTopics[i] = []common.Hash{t} + } + } else { + fmt.Errorf("topic[%d][%d] not a string", i, j) + } + } + } else { + return fmt.Errorf("topic[%d] invalid", i) + } + } + args.Topics = parsedTopics + } + } + + return nil +} + +func (s *FilterService) NewFilter(args NewFilterArgs) (string, error) { + externalId, err := newFilterId() + if err != nil { + return "", err + } + + var id int + if len(args.Addresses) > 0 { + id = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics) + } else { + id = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics) + } + + s.filterMapMu.Lock() + s.filterMapping[externalId] = id + s.filterMapMu.Unlock() + + return externalId, nil +} + +func (s *FilterService) GetLogs(args NewFilterArgs) (vm.Logs) { + filter := New(s.chainDb) + filter.SetBeginBlock(args.FromBlock.Int64()) + filter.SetEndBlock(args.ToBlock.Int64()) + filter.SetAddresses(args.Addresses) + filter.SetTopics(args.Topics) + + return filter.Find() +} + +func (s *FilterService) UninstallFilter(filterId string) bool { + s.filterMapMu.Lock() + defer s.filterMapMu.Unlock() + + id, ok := s.filterMapping[filterId] + if !ok { + return false + } + + defer s.filterManager.Remove(id) + delete(s.filterMapping, filterId) + + if _, ok := s.logQueue[id]; ok { + s.logMu.Lock() + defer s.logMu.Unlock() + delete(s.logQueue, id) + return true + } + if _, ok := s.blockQueue[id]; ok { + s.blockMu.Lock() + defer s.blockMu.Unlock() + delete(s.blockQueue, id) + return true + } + if _, ok := s.transactionQueue[id]; ok { + s.transactionMu.Lock() + defer s.transactionMu.Unlock() + delete(s.transactionQueue, id) + return true + } + + return false +} + +func (s *FilterService) getFilterType(id int) byte { + if _, ok := s.blockQueue[id]; ok { + return blockFilterTy + } else if _, ok := s.transactionQueue[id]; ok { + return transactionFilterTy + } else if _, ok := s.logQueue[id]; ok { + return logFilterTy + } + + return unknownFilterTy +} + +func (s *FilterService) blockFilterChanged(id int) []common.Hash { + s.blockMu.Lock() + defer s.blockMu.Unlock() + + if s.blockQueue[id] != nil { + return s.blockQueue[id].get() + } + return []common.Hash{} +} + +func (s *FilterService) transactionFilterChanged(id int) []common.Hash { + s.blockMu.Lock() + defer s.blockMu.Unlock() + + if s.transactionQueue[id] != nil { + return s.transactionQueue[id].get() + } + return []common.Hash{} +} + +func (s *FilterService) logFilterChanged(id int) vm.Logs { + s.logMu.Lock() + defer s.logMu.Unlock() + + if s.logQueue[id] != nil { + return s.logQueue[id].get() + } + return vm.Logs{} +} + +func (s *FilterService) GetFilterLogs(filterId string) vm.Logs { + id, ok := s.filterMapping[filterId] + if !ok { + return vm.Logs{} + } + + if filter := s.filterManager.Get(id); filter != nil { + return filter.Find() + } + + return vm.Logs{} +} + +func (s *FilterService) GetFilterChanges(filterId string) interface{} { + s.filterMapMu.Lock() + id, ok := s.filterMapping[filterId] + s.filterMapMu.Unlock() + + if !ok { // filter not found + return nil + } + + switch s.getFilterType(id) { + case blockFilterTy: + return s.blockFilterChanged(id) + case transactionFilterTy: + return s.transactionFilterChanged(id) + case logFilterTy: + return s.logFilterChanged(id) + } + + return nil +} + +type logQueue struct { + mu sync.Mutex + + logs vm.Logs + timeout time.Time + id int +} + +func (l *logQueue) add(logs ...*vm.Log) { + l.mu.Lock() + defer l.mu.Unlock() + + l.logs = append(l.logs, logs...) +} + +func (l *logQueue) get() vm.Logs { + l.mu.Lock() + defer l.mu.Unlock() + + l.timeout = time.Now() + tmp := l.logs + l.logs = nil + return tmp +} + +type hashQueue struct { + mu sync.Mutex + + hashes []common.Hash + timeout time.Time + id int +} + +func (l *hashQueue) add(hashes ...common.Hash) { + l.mu.Lock() + defer l.mu.Unlock() + + l.hashes = append(l.hashes, hashes...) +} + +func (l *hashQueue) get() []common.Hash { + l.mu.Lock() + defer l.mu.Unlock() + + l.timeout = time.Now() + tmp := l.hashes + l.hashes = nil + return tmp +} + +func newFilterId() (string, error) { + var subid [16]byte + n, _ := rand.Read(subid[:]) + if n != 16 { + return "", errors.New("Unable to generate filter id") + } + return "0x" + hex.EncodeToString(subid[:]), nil +} diff --git a/eth/web3_rpc.go b/eth/web3_rpc.go new file mode 100644 index 0000000000..4919941c2c --- /dev/null +++ b/eth/web3_rpc.go @@ -0,0 +1,22 @@ +package eth + +import ( + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" +) + +type Web3Service struct { + e *Ethereum +} + +func NewWeb3Service(e *Ethereum) *Web3Service { + return &Web3Service{e} +} + +func (s *Web3Service) ClientVersion() string { + return s.e.ClientVersion() +} + +func (s *Web3Service) Sha3(data string) string { + return common.ToHex(crypto.Sha3(common.FromHex(data))) +} \ No newline at end of file diff --git a/ethdb/interface.go b/ethdb/interface.go index a6a3c31ff9..f4b787a52a 100644 --- a/ethdb/interface.go +++ b/ethdb/interface.go @@ -28,4 +28,3 @@ type Batch interface { Put(key, value []byte) error Write() error } - diff --git a/miner/miner_rpc.go b/miner/miner_rpc.go new file mode 100644 index 0000000000..d95b0cbd30 --- /dev/null +++ b/miner/miner_rpc.go @@ -0,0 +1,52 @@ +package miner + +import ( + "github.com/ethereum/go-ethereum/common" + rpc "github.com/ethereum/go-ethereum/rpc/v2" + "fmt" + "github.com/ethereum/go-ethereum/logger/glog" +) + +type MinerService struct { + miner *Miner + agent *RemoteAgent +} + +func NewMinerService(miner *Miner) *MinerService { + return &MinerService{miner, NewRemoteAgent()} +} + +// Mining returns an indication if this node is currently mining. +func (s *MinerService) Mining() bool { + return s.miner.Mining() +} + +// SubmitWork can be used by external miner to submit their POW solution. It returns an indication if the work was +// accepted. Note, this is not an indication if the provided work was valid! +func (s *MinerService) SubmitWork(nonce rpc.HexNumber, solution, digest common.Hash) bool { + return s.agent.SubmitWork(nonce.Uint64(), digest, solution) +} + +// GetWork returns a work package for external miner. The work package consists of 3 strings +// result[0], 32 bytes hex encoded current block header pow-hash +// result[1], 32 bytes hex encoded seed hash used for DAG +// result[2], 32 bytes hex encoded boundary condition ("target"), 2^256/difficulty +func (s *MinerService) GetWork() ([]string, error) { + if !s.Mining() { + s.miner.Start(s.miner.coinbase, 0) + } + if work, err := s.agent.GetWork(); err == nil { + return work[:], nil + } else { + glog.Infof("%v\n", err) + } + return nil, fmt.Errorf("mining not ready") +} + +// SubmitHashrate can be used for remote miners to submit their hash rate. This enables the node to report the combined +// hash rate of all miners which submit work through this node. It accepts the miner hash rate and an identifier which +// must be unique between nodes. +func (s *MinerService) SubmitHashrate(hashrate rpc.HexNumber, id common.Hash) bool { + s.agent.SubmitHashrate(id, hashrate.Uint64()) + return true +} diff --git a/miner/worker.go b/miner/worker.go index 754a6fc483..1a411ae20b 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -327,6 +327,7 @@ func (self *worker) wait() { go func(block *types.Block, logs vm.Logs, receipts []*types.Receipt) { self.mux.Post(core.NewMinedBlockEvent{block}) self.mux.Post(core.ChainEvent{block, block.Hash(), logs}) + if stat == core.CanonStatTy { self.mux.Post(core.ChainHeadEvent{block}) self.mux.Post(logs) diff --git a/p2p/p2p_rpc.go b/p2p/p2p_rpc.go new file mode 100644 index 0000000000..117052095f --- /dev/null +++ b/p2p/p2p_rpc.go @@ -0,0 +1,29 @@ +package p2p + +import ( + rpc "github.com/ethereum/go-ethereum/rpc/v2" + "fmt" +) +type NetService struct { + net *Server + networkVersion int +} + +func NewNetService(net *Server, networkVersion int) *NetService { + return &NetService{net, networkVersion} +} + +// Listening returns an indication if the node is listening for network connections. +func (s *NetService) Listening() bool { + return true // always listening +} + +// Peercount returns the number of connected peers +func (s *NetService) PeerCount() *rpc.HexNumber { + return rpc.NewHexNumber(s.net.PeerCount()) +} + +// ProtocolVersion returns the current ethereum protocol version. +func (s *NetService) Version() string { + return fmt.Sprintf("%d", s.networkVersion) +} diff --git a/rpc/api/mergedapi.go b/rpc/api/mergedapi.go index 8f4ef8e603..92e1e2bb77 100644 --- a/rpc/api/mergedapi.go +++ b/rpc/api/mergedapi.go @@ -39,9 +39,11 @@ func newMergedApi(apis ...shared.EthereumApi) *MergedApi { mergedApi.methods = make(map[string]shared.EthereumApi) for _, api := range apis { - mergedApi.apis[api.Name()] = api.ApiVersion() - for _, method := range api.Methods() { - mergedApi.methods[method] = api + if api != nil { + mergedApi.apis[api.Name()] = api.ApiVersion() + for _, method := range api.Methods() { + mergedApi.methods[method] = api + } } } return mergedApi diff --git a/rpc/api/utils.go b/rpc/api/utils.go index 8351e88d39..c94ed90ea7 100644 --- a/rpc/api/utils.go +++ b/rpc/api/utils.go @@ -184,6 +184,8 @@ func ParseApiString(apistr string, codec codec.Codec, xeth *xeth.XEth, eth *eth. apis[i] = NewPersonalApi(xeth, eth, codec) case shared.Web3ApiName: apis[i] = NewWeb3Api(xeth, codec) + case "rpc": // gives information about the RPC interface + continue default: return nil, fmt.Errorf("Unknown API '%s'", name) } diff --git a/rpc/comms/ipc.go b/rpc/comms/ipc.go index 882d62ab49..67edbe9aea 100644 --- a/rpc/comms/ipc.go +++ b/rpc/comms/ipc.go @@ -69,13 +69,29 @@ func (self *ipcClient) SupportedModules() (map[string]string, error) { req := shared.Request{ Id: 1, Jsonrpc: "2.0", - Method: "modules", + Method: "rpc_modules", } if err := self.coder.WriteResponse(req); err != nil { return nil, err } + res, _ := self.coder.ReadResponse() + if sucRes, ok := res.(*shared.SuccessResponse); ok { + data, _ := json.Marshal(sucRes.Result) + modules := make(map[string]string) + if err := json.Unmarshal(data, &modules); err == nil { + return modules, nil + } + } + + + // old version uses modules instead of rpc_modules, this can be removed after full migration + req.Method = "modules" + if err := self.coder.WriteResponse(req); err != nil { + return nil, err + } + res, err := self.coder.ReadResponse() if err != nil { return nil, err diff --git a/rpc/v2/json.go b/rpc/v2/json.go index e8ba340945..fc0e03b863 100644 --- a/rpc/v2/json.go +++ b/rpc/v2/json.go @@ -18,10 +18,13 @@ package v2 import ( "encoding/json" + "fmt" "io" "reflect" "strings" "sync/atomic" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/logger" ) const ( @@ -131,16 +134,17 @@ func parseRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, RPCError) { if in.Method == subscribeMethod { reqs := []rpcRequest{rpcRequest{id: in.Id, isPubSub: true}} if len(in.Payload) > 0 { - // first param must be service.method name - subscriptionMethod := []reflect.Type{reflect.TypeOf("")} - if args, err := parsePositionalArguments(in.Payload, subscriptionMethod); err == nil && len(args) == 1 { - elems := strings.Split(args[0].String(), serviceMethodSeparator) - if len(elems) == 2 { - reqs[0].service, reqs[0].method = elems[0], elems[1] - reqs[0].params = in.Payload - return reqs, false, nil - } + // first param must be subscription name + var subscribeMethod [1]string + if err := json.Unmarshal(in.Payload, &subscribeMethod); err != nil { + glog.V(logger.Debug).Infof("Unable to parse subscription method: %v\n", err) + return nil, false, &invalidRequestError{"Unable to parse subscription request"} } + + // all subscriptions are made on the eth service + reqs[0].service, reqs[0].method = "eth", subscribeMethod[0] + reqs[0].params = in.Payload + return reqs, false, nil } return nil, false, &invalidRequestError{"Unable to parse subscription request"} } @@ -177,16 +181,16 @@ func parseBatchRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, RPCErro if r.Method == subscribeMethod { requests[i] = rpcRequest{id: r.Id, isPubSub: true} if len(r.Payload) > 0 { - // first param must be service.method name - subscriptionMethod := []reflect.Type{reflect.TypeOf("")} - if args, err := parsePositionalArguments(r.Payload, subscriptionMethod); err == nil && len(args) == 1 { - elems := strings.Split(args[0].String(), serviceMethodSeparator) - if len(elems) == 2 { - requests[i].service, requests[i].method = elems[0], elems[1] - requests[i].params = r.Payload - continue - } + var subscribeMethod [1]string + if err := json.Unmarshal(r.Payload, &subscribeMethod); err != nil { + glog.V(logger.Debug).Infof("Unable to parse subscription method: %v\n", err) + return nil, false, &invalidRequestError{"Unable to parse subscription request"} } + + // all subscriptions are made on the eth service + requests[i].service, requests[i].method = "eth", subscribeMethod[0] + requests[i].params = r.Payload + continue } return nil, true, &invalidRequestError{"Unable to parse (un)subscribe request arguments"} @@ -222,20 +226,48 @@ func (c *jsonCodec) ParseRequestArguments(argTypes []reflect.Type, params interf } } +func countArguments(args json.RawMessage) (int, error) { + var cnt []interface{} + if err := json.Unmarshal(args, &cnt); err != nil { + return -1, nil + } + return len(cnt), nil +} + // parsePositionalArguments tries to parse the given args to an array of values with the given types. It returns the // parsed values or an error when the args could not be parsed. func parsePositionalArguments(args json.RawMessage, argTypes []reflect.Type) ([]reflect.Value, RPCError) { argValues := make([]reflect.Value, len(argTypes)) params := make([]interface{}, len(argTypes)) + + n, err := countArguments(args) + if err != nil { + return nil, &invalidParamsError{err.Error()} + } + if n != len(argTypes) { + return nil, &invalidParamsError{fmt.Sprintf("insufficient params, want %d have %d", len(argTypes), n)} + + } + for i, t := range argTypes { if t.Kind() == reflect.Ptr { // values must be pointers for the Unmarshal method, reflect. // Dereference otherwise reflect.New would create **SomeType argValues[i] = reflect.New(t.Elem()) params[i] = argValues[i].Interface() + + // when not specified blockNumbers are by default latest (-1) + if blockNumber, ok := params[i].(*BlockNumber); ok { + *blockNumber = BlockNumber(-1) + } } else { argValues[i] = reflect.New(t) params[i] = argValues[i].Interface() + + // when not specified blockNumbers are by default latest (-1) + if blockNumber, ok := params[i].(*BlockNumber); ok { + *blockNumber = BlockNumber(-1) + } } } @@ -255,6 +287,9 @@ func parsePositionalArguments(args json.RawMessage, argTypes []reflect.Type) ([] // CreateResponse will create a JSON-RPC success response with the given id and reply as result. func (c *jsonCodec) CreateResponse(id int64, reply interface{}) interface{} { + if isHexNum(reflect.TypeOf(reply)) { + return &jsonSuccessResponse{Version: jsonRPCVersion, Id: id, Result: fmt.Sprintf(`%#x`, reply)} + } return &jsonSuccessResponse{Version: jsonRPCVersion, Id: id, Result: reply} } @@ -272,6 +307,11 @@ func (c *jsonCodec) CreateErrorResponseWithInfo(id *int64, err RPCError, info in // CreateNotification will create a JSON-RPC notification with the given subscription id and event as params. func (c *jsonCodec) CreateNotification(subid string, event interface{}) interface{} { + if isHexNum(reflect.TypeOf(event)) { + return &jsonNotification{Version: jsonRPCVersion, Method: notificationMethod, + Params: jsonSubscription{Subscription: subid, Result: fmt.Sprintf(`%#x`, event)}} + } + return &jsonNotification{Version: jsonRPCVersion, Method: notificationMethod, Params: jsonSubscription{Subscription: subid, Result: event}} } diff --git a/rpc/v2/server.go b/rpc/v2/server.go index 6f6f6af19c..e1e0fa3a6a 100644 --- a/rpc/v2/server.go +++ b/rpc/v2/server.go @@ -23,11 +23,32 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" + "runtime" ) // NewServer will create a new server instance with no registered handlers. func NewServer() *Server { - return &Server{services: make(serviceRegistry), subscriptions: make(subscriptionRegistry)} + server := &Server{services: make(serviceRegistry), subscriptions: make(subscriptionRegistry)} + + rpcService := &RPCService{server} + server.RegisterName("rpc", rpcService) + + return server +} + +// RPCService gives meta information about the server. +// e.g. gives information about the loaded modules. +type RPCService struct { + server *Server +} + +// Modules returns the list of RPC services with their version number +func (s *RPCService) Modules() map[string]string { + modules := make(map[string]string) + for name, _ := range s.server.services { + modules[name] = "1.0" + } + return modules } // Register publishes suitable methods of rcvr. Methods will be published @@ -54,9 +75,9 @@ func (s *Server) register(rcvr interface{}, name string, useName bool) error { svc := new(service) svc.typ = reflect.TypeOf(rcvr) - svc.rcvr = reflect.ValueOf(rcvr) + rcvrVal := reflect.ValueOf(rcvr) - sname := reflect.Indirect(svc.rcvr).Type().Name() + sname := reflect.Indirect(rcvrVal).Type().Name() if useName { sname = name } @@ -67,12 +88,25 @@ func (s *Server) register(rcvr interface{}, name string, useName bool) error { return fmt.Errorf("%s is not exported", sname) } - if _, present := s.services[sname]; present { - return fmt.Errorf("%s already registered", sname) + // already a previous service register under given sname, merge methods/subscriptions + if regsvc, present := s.services[sname]; present { + methods, subscriptions := suitableCallbacks(rcvrVal, svc.typ) + if len(methods) == 0 && len(subscriptions) == 0 { + return fmt.Errorf("Service doesn't have any suitable methods/subscriptions to expose") + } + + for _, m := range methods { + regsvc.callbacks[formatName(m.method.Name)] = m + } + for _, s := range subscriptions { + regsvc.subscriptions[formatName(s.method.Name)] = s + } + + return nil } svc.name = sname - svc.callbacks, svc.subscriptions = suitableCallbacks(svc.typ) + svc.callbacks, svc.subscriptions = suitableCallbacks(rcvrVal, svc.typ) if len(svc.callbacks) == 0 && len(svc.subscriptions) == 0 { return fmt.Errorf("Service doesn't have any suitable methods/subscriptions to expose") @@ -91,7 +125,15 @@ func (s *Server) register(rcvr interface{}, name string, useName bool) error { // 2. supports notifications (pub/sub) // 3. supports request batches func (s *Server) ServeCodec(codec ServerCodec) { - defer codec.Close() + defer func() { + if err := recover(); err != nil { + const size = 64 << 10 + buf := make([]byte, size) + buf = buf[:runtime.Stack(buf, false)] + glog.Errorln(buf) + } + codec.Close() + }() for { reqs, batch, err := s.readRequest(codec) @@ -113,28 +155,7 @@ func (s *Server) ServeCodec(codec ServerCodec) { // It will then send the notification to the client, when it fails the codec is closed. When the event has multiple // fields an array of values is returned. func sendNotification(codec ServerCodec, subid string, event interface{}) { - typ := reflect.TypeOf(event) - val := reflect.ValueOf(event) - for typ.Kind() == reflect.Ptr { - typ = typ.Elem() - val = val.Elem() - } - - fields := make(map[string]interface{}) - var notification interface{} - - if typ.Kind() == reflect.Struct { - if typ.NumField() == 1 { - notification = codec.CreateNotification(subid, val.Field(0).Interface()) - } else { - for i := 0; i < typ.NumField(); i++ { - fields[typ.Field(i).Name] = val.Field(i).Interface() - } - notification = codec.CreateNotification(subid, fields) - } - } else { - notification = codec.CreateNotification(subid, val.Interface()) - } + notification := codec.CreateNotification(subid, event) if err := codec.Write(notification); err != nil { codec.Close() @@ -146,7 +167,7 @@ func sendNotification(codec ServerCodec, subid string, event interface{}) { // 2. create a notification of the event and send it the client when it matches the criteria // It will unsubscribe the subscription when the socket is closed or the subscription is unsubscribed by the user. func (s *Server) createSubscription(c ServerCodec, req *serverRequest) (string, error) { - args := []reflect.Value{req.rcvr} + args := []reflect.Value{req.callb.rcvr} if len(req.args) > 0 { args = append(args, req.args...) } @@ -176,7 +197,7 @@ func (s *Server) createSubscription(c ServerCodec, req *serverRequest) (string, if recvOk { // send notification if event, ok := notification.Interface().(*event.Event); ok { if subscription.match == nil || subscription.match(event.Data) { - sendNotification(c, subid, event.Data) + sendNotification(c, subid, subscription.format(event.Data)) } } } else { // user send an eth_unsubscribe request @@ -274,7 +295,7 @@ func (s *Server) exec(codec ServerCodec, req *serverRequest) { return } - arguments := []reflect.Value{req.rcvr} + arguments := []reflect.Value{req.callb.rcvr} if len(req.args) > 0 { arguments = append(arguments, req.args...) } @@ -353,7 +374,7 @@ func (s *Server) execBatch(codec ServerCodec, requests []*serverRequest) { continue } - arguments := []reflect.Value{req.rcvr} + arguments := []reflect.Value{req.callb.rcvr} if len(req.args) > 0 { arguments = append(arguments, req.args...) } @@ -417,7 +438,7 @@ func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, RPCErro } if callb, ok := svc.subscriptions[r.method]; ok { - requests[i] = &serverRequest{id: r.id, svcname: svc.name, rcvr: svc.rcvr, callb: callb} + requests[i] = &serverRequest{id: r.id, svcname: svc.name, callb: callb} if r.params != nil && len(callb.argTypes) > 0 { argTypes := []reflect.Type{reflect.TypeOf("")} argTypes = append(argTypes, callb.argTypes...) @@ -431,7 +452,7 @@ func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, RPCErro } if callb, ok := svc.callbacks[r.method]; ok { - requests[i] = &serverRequest{id: r.id, svcname: svc.name, rcvr: svc.rcvr, callb: callb} + requests[i] = &serverRequest{id: r.id, svcname: svc.name, callb: callb} if r.params != nil && len(callb.argTypes) > 0 { if args, err := codec.ParseRequestArguments(callb.argTypes, r.params); err == nil { requests[i].args = args diff --git a/rpc/v2/server_test.go b/rpc/v2/server_test.go index 3ff22fdd71..1ca954923f 100644 --- a/rpc/v2/server_test.go +++ b/rpc/v2/server_test.go @@ -38,10 +38,6 @@ func TestServerRegister(t *testing.T) { t.Fatalf("%v", err) } - if err := server.RegisterName("calc", service); err == nil { - t.Fatal("Second time registering the same service should raise an error") - } - if len(server.services) != 1 { t.Fatalf("Expected 1 service entry but got %d", len(server.services)) } diff --git a/rpc/v2/types.go b/rpc/v2/types.go index a9286e2c6b..38fd5f4d5f 100644 --- a/rpc/v2/types.go +++ b/rpc/v2/types.go @@ -17,7 +17,11 @@ package v2 import ( + "fmt" + "math" + "math/big" "reflect" + "strings" "sync" @@ -26,6 +30,7 @@ import ( // callback is a method callback which was registered in the server type callback struct { + rcvr reflect.Value // receiver of method method reflect.Method // callback argTypes []reflect.Type // input argument types errPos int // err return idx, of -1 when method cannot return error @@ -108,21 +113,35 @@ type ServerCodec interface { // SubscriptionMatcher returns true if the given value matches the criteria specified by the user type SubscriptionMatcher func(interface{}) bool +// SubscriptionOutputFormat accepts event data and has the ability to format the data before it is send to the client +type SubscriptionOutputFormat func(interface{}) interface{} + +// defaultSubscriptionOutputFormatter returns data and is used as default output format for notifications +func defaultSubscriptionOutputFormatter(data interface{}) interface{} { + return data +} + // Subscription is used by the server to send notifications to the client type Subscription struct { - sub event.Subscription - match SubscriptionMatcher + sub event.Subscription + match SubscriptionMatcher + format SubscriptionOutputFormat } // NewSubscription create a new RPC subscription func NewSubscription(sub event.Subscription) Subscription { - return Subscription{sub, nil} + return Subscription{sub, nil, defaultSubscriptionOutputFormatter} +} + +// NewSubscriptionWithOutputFormat create a new RPC subscription which a custom notification output format +func NewSubscriptionWithOutputFormat(sub event.Subscription, formatter SubscriptionOutputFormat) Subscription { + return Subscription{sub, nil, formatter} } // NewSubscriptionFiltered will create a new subscription. For each raised event the given matcher is // called. If it returns true the event is send as notification to the client, otherwise it is ignored. func NewSubscriptionFiltered(sub event.Subscription, match SubscriptionMatcher) Subscription { - return Subscription{sub, match} + return Subscription{sub, match, defaultSubscriptionOutputFormatter} } // Chan returns the channel where new events will be published. It's up the user to call the matcher to @@ -135,3 +154,186 @@ func (s *Subscription) Chan() <-chan *event.Event { func (s *Subscription) Unsubscribe() { s.sub.Unsubscribe() } + +// HexNumber serializes a number to hex format using the "%#x" format +type HexNumber big.Int + +// NewHexNumber creates a new hex number instance which will serialize the given val with `%#x` on marshal. +func NewHexNumber(val interface{}) *HexNumber { + if val == nil { + return nil + } + + if v, ok := val.(*big.Int); ok && v != nil { + hn := new(big.Int).Set(v) + return (*HexNumber)(hn) + } + + rval := reflect.ValueOf(val) + + var unsigned uint64 + utype := reflect.TypeOf(unsigned) + if t := rval.Type(); t.ConvertibleTo(utype) { + hn := new(big.Int).SetUint64(rval.Convert(utype).Uint()) + return (*HexNumber)(hn) + } + + var signed int64 + stype := reflect.TypeOf(signed) + if t := rval.Type(); t.ConvertibleTo(stype) { + hn := new(big.Int).SetInt64(rval.Convert(stype).Int()) + return (*HexNumber)(hn) + } + + return nil +} + +func (h *HexNumber) UnmarshalJSON(input []byte) error { + length := len(input) + if length >= 2 && input[0] == '"' && input[length - 1] == '"' { + input = input[1 : length - 1] + } + + hn := (*big.Int)(h) + if _, ok := hn.SetString(string(input), 0); ok { + return nil + } + + return fmt.Errorf("Unable to parse number") +} + +// MarshalJSON serialize the hex number instance to a hex representation. +func (h *HexNumber) MarshalJSON() ([]byte, error) { + if h != nil { + hn := (*big.Int)(h) + if hn.BitLen() == 0 { + return []byte(`"0x0"`), nil + } + return []byte(fmt.Sprintf(`"0x%x"`, hn)), nil + } + return nil, nil +} + +func (h *HexNumber) Int() int { + hn := (*big.Int)(h) + return int(hn.Int64()) +} + +func (h *HexNumber) Int64() int64 { + hn := (*big.Int)(h) + return hn.Int64() +} + +func (h *HexNumber) Uint() uint { + hn := (*big.Int)(h) + return uint(hn.Uint64()) +} + +func (h *HexNumber) Uint64() uint64 { + hn := (*big.Int)(h) + return hn.Uint64() +} + +func (h *HexNumber) BigInt() *big.Int { + return (*big.Int)(h) +} + +type Number int64 +func (n *Number) UnmarshalJSON(data []byte) error { + input := strings.TrimSpace(string(data)) + + if len(input) >= 2 && input[0] == '"' && input[len(input) - 1] == '"' { + input = input[1 : len(input) - 1] + } + + if len(input) == 0 { + *n = Number(latestBlockNumber.Int64()) + return nil + } + + in := new(big.Int) + _, ok := in.SetString(input, 0) + + if !ok { // test if user supplied string tag + return fmt.Errorf(`invalid number %s`, data) + } + + if in.Cmp(earliestBlockNumber) >= 0 && in.Cmp(maxBlockNumber) <= 0 { + *n = Number(in.Int64()) + return nil + } + + return fmt.Errorf("blocknumber not in range [%d, %d]", earliestBlockNumber, maxBlockNumber) +} + +func (n *Number) Int64() int64 { + return *(*int64)(n) +} + +var ( + pendingBlockNumber = big.NewInt(-2) + latestBlockNumber = big.NewInt(-1) + earliestBlockNumber = big.NewInt(0) + maxBlockNumber = big.NewInt(math.MaxInt64) +) + +type BlockNumber int64 + +const ( + PendingBlockNumber = BlockNumber(-2) + LatestBlockNumber = BlockNumber(-1) +) + +// UnmarshalJSON parses the given JSON fragement into a BlockNumber. It supports: +// - "latest" or "earliest" as string arguments +// - the block number +// Returned errors: +// - an unsupported error when "pending" is specified (not yet implemented) +// - an invalid block number error when the given argument isn't a known strings +// - an out of range error when the given block number is either too little or too large +func (bn *BlockNumber) UnmarshalJSON(data []byte) error { + input := strings.TrimSpace(string(data)) + + if len(input) >= 2 && input[0] == '"' && input[len(input) - 1] == '"' { + input = input[1 : len(input) - 1] + } + + if len(input) == 0 { + *bn = BlockNumber(latestBlockNumber.Int64()) + return nil + } + + in := new(big.Int) + _, ok := in.SetString(input, 0) + + if !ok { // test if user supplied string tag + strBlockNumber := input + if strBlockNumber == "latest" { + *bn = BlockNumber(latestBlockNumber.Int64()) + return nil + } + + if strBlockNumber == "earliest" { + *bn = BlockNumber(earliestBlockNumber.Int64()) + return nil + } + + if strBlockNumber == "pending" { + *bn = BlockNumber(pendingBlockNumber.Int64()) + return nil + } + + return fmt.Errorf(`invalid blocknumber %s`, data) + } + + if in.Cmp(earliestBlockNumber) >= 0 && in.Cmp(maxBlockNumber) <= 0 { + *bn = BlockNumber(in.Int64()) + return nil + } + + return fmt.Errorf("blocknumber not in range [%d, %d]", earliestBlockNumber, maxBlockNumber) +} + +func (bn *BlockNumber) Int64() int64 { + return (int64)(*bn) +} diff --git a/rpc/v2/types_test.go b/rpc/v2/types_test.go new file mode 100644 index 0000000000..f73a2369ee --- /dev/null +++ b/rpc/v2/types_test.go @@ -0,0 +1,57 @@ +package v2 + +import ( + "bytes" + "encoding/json" + "math/big" + "testing" +) + +func TestNewHexNumber(t *testing.T) { + tests := []interface{}{big.NewInt(123), int64(123), uint64(123), int8(123), uint8(123)} + + for i, v := range tests { + hn := NewHexNumber(v) + if hn == nil { + t.Fatalf("Unable to create hex number instance for tests[%d]", i) + } + if hn.Int64() != 123 { + t.Fatalf("expected %d, got %d on value tests[%d]", 123, hn.Int64(), i) + } + } + + failures := []interface{}{"", nil, []byte{1, 2, 3, 4}} + for i, v := range failures { + hn := NewHexNumber(v) + if hn != nil { + t.Fatalf("Creating a nex number instance of %T should fail (failures[%d])", failures[i], i) + } + } +} + +func TestHexNumberUnmarshalJSON(t *testing.T) { + tests := []string{`"0x4d2"`, "1234", `"1234"`} + for i, v := range tests { + var hn HexNumber + if err := json.Unmarshal([]byte(v), &hn); err != nil { + t.Fatalf("Test %d failed - %s", i, err) + } + + if hn.Int64() != 1234 { + t.Fatalf("Expected %d, got %d for test[%d]", 1234, hn.Int64(), i) + } + } +} + +func TestHexNumberMarshalJSON(t *testing.T) { + hn := NewHexNumber(1234567890) + got, err := json.Marshal(hn) + if err != nil { + t.Fatalf("Unable to marshal hex number - %s", err) + } + + exp := []byte(`"0x499602d2"`) + if bytes.Compare(exp, got) != 0 { + t.Fatalf("Invalid json.Marshal, expected '%s', got '%s'", exp, got) + } +} diff --git a/rpc/v2/utils.go b/rpc/v2/utils.go index ee6ee777d0..7b7006e8a4 100644 --- a/rpc/v2/utils.go +++ b/rpc/v2/utils.go @@ -20,6 +20,7 @@ import ( "crypto/rand" "encoding/hex" "errors" + "math/big" "reflect" "unicode" "unicode/utf8" @@ -70,10 +71,48 @@ func isPubSub(methodType reflect.Type) bool { return isSubscriptionType(methodType.Out(0)) && isErrorType(methodType.Out(1)) } +// formatName will convert to first character to lower case +func formatName(name string) string { + ret := []rune(name) + if len(ret) > 0 { + ret[0] = unicode.ToLower(ret[0]) + } + return string(ret) +} + +var bigIntType = reflect.TypeOf((*big.Int)(nil)).Elem() + +// Indication if this type should be serialized in hex +func isHexNum(t reflect.Type) bool { + if t == nil { + return false + } + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + + return t == bigIntType +} + +var blockNumberType = reflect.TypeOf((*BlockNumber)(nil)).Elem() + +// Indication if the given block is a BlockNumber +func isBlockNumber(t reflect.Type) bool { + if t == nil { + return false + } + + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + + return t == blockNumberType +} + // suitableCallbacks iterates over the methods of the given type. It will determine if a method satisfies the criteria // for a RPC callback or a subscription callback and adds it to the collection of callbacks or subscriptions. See server // documentation for a summary of these criteria. -func suitableCallbacks(typ reflect.Type) (callbacks, subscriptions) { +func suitableCallbacks(rcvr reflect.Value, typ reflect.Type) (callbacks, subscriptions) { callbacks := make(callbacks) subscriptions := make(subscriptions) @@ -81,13 +120,14 @@ METHODS: for m := 0; m < typ.NumMethod(); m++ { method := typ.Method(m) mtype := method.Type - mname := method.Name + mname := formatName(method.Name) if method.PkgPath != "" { // method must be exported continue } if isPubSub(mtype) { var h callback + h.rcvr = rcvr h.method = method h.errPos = -1 h.isSubscribe = true @@ -107,6 +147,7 @@ METHODS: } var h callback + h.rcvr = rcvr h.method = method numIn := mtype.NumIn()