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/geth/main.go b/cmd/geth/main.go index 8798bbb273..cb90f43f6d 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -313,6 +313,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso utils.IPCDisabledFlag, utils.IPCApiFlag, utils.IPCPathFlag, + utils.IPCExperimental, utils.ExecFlag, utils.WhisperEnabledFlag, utils.DevModeFlag, @@ -398,7 +399,7 @@ func attach(ctx *cli.Context) { client, err = comms.ClientFromEndpoint(ctx.Args().First(), codec.JSON) } else { cfg := comms.IpcConfig{ - Endpoint: utils.IpcSocketPath(ctx), + Endpoint: (&node.Config{DataDir: ctx.GlobalString(utils.DataDirFlag.Name), IpcPath: ctx.GlobalString(utils.IPCPathFlag.Name)}).IpcEndpoint(), } client, err = comms.NewIpcClient(cfg, codec.JSON) } @@ -506,11 +507,6 @@ func startNode(ctx *cli.Context, stack *node.Node) { } } // Start auxiliary services if enabled. - if !ctx.GlobalBool(utils.IPCDisabledFlag.Name) { - if err := utils.StartIPC(stack, ctx); err != nil { - utils.Fatalf("Failed to start IPC: %v", err) - } - } if ctx.GlobalBool(utils.RPCEnabledFlag.Name) { if err := utils.StartRPC(stack, ctx); err != nil { utils.Fatalf("Failed to start RPC: %v", err) @@ -534,7 +530,7 @@ func accountList(ctx *cli.Context) { } } -// getPassPhrase retrieves the passwor associated with an account, either fetched +// getPassPhrase retrieves the password associated with an account, either fetched // from a list of preloaded passphrases, or requested interactively from the user. func getPassPhrase(prompt string, confirmation bool, i int, passwords []string) string { // If a list of passwords was supplied, retrieve from them diff --git a/cmd/geth/monitorcmd.go b/cmd/geth/monitorcmd.go index a45d29b8f2..9537596b16 100644 --- a/cmd/geth/monitorcmd.go +++ b/cmd/geth/monitorcmd.go @@ -27,7 +27,6 @@ import ( "github.com/codegangsta/cli" "github.com/ethereum/go-ethereum/cmd/utils" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/comms" @@ -37,7 +36,7 @@ import ( var ( monitorCommandAttachFlag = cli.StringFlag{ Name: "attach", - Value: "ipc:" + common.DefaultIpcPath(), + Value: "ipc:" + utils.IPCPathFlag.Value.Value, Usage: "API endpoint to attach to", } monitorCommandRowsFlag = cli.IntFlag{ diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go index 5c09e29ce4..7a6ff704c4 100644 --- a/cmd/geth/usage.go +++ b/cmd/geth/usage.go @@ -158,6 +158,7 @@ var AppHelpFlagGroups = []flagGroup{ Flags: []cli.Flag{ utils.WhisperEnabledFlag, utils.NatspecEnabledFlag, + utils.IPCExperimental, }, }, { diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 839ec3f027..f8f4a76f88 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -23,7 +23,6 @@ import ( "log" "math" "math/big" - "net" "net/http" "os" "path/filepath" @@ -52,8 +51,6 @@ import ( "github.com/ethereum/go-ethereum/rpc/api" "github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/comms" - "github.com/ethereum/go-ethereum/rpc/shared" - "github.com/ethereum/go-ethereum/rpc/useragent" "github.com/ethereum/go-ethereum/whisper" "github.com/ethereum/go-ethereum/xeth" ) @@ -297,8 +294,12 @@ var ( } IPCPathFlag = DirectoryFlag{ Name: "ipcpath", - Usage: "Filename for IPC socket/pipe", - Value: DirectoryString{common.DefaultIpcPath()}, + Usage: "Filename for IPC socket/pipe within the datadir (explicit paths escape it)", + Value: DirectoryString{"geth.ipc"}, + } + IPCExperimental = cli.BoolFlag{ + Name: "ipcexp", + Usage: "Enable the new RPC implementation", } ExecFlag = cli.StringFlag{ Name: "exec", @@ -405,6 +406,15 @@ func MustMakeDataDir(ctx *cli.Context) string { return "" } +// MakeIpcPath creates an IPC path configuration from the set command line flags, +// returning an empty string if IPC was explicitly disabled, or the set path. +func MakeIpcPath(ctx *cli.Context) string { + if ctx.GlobalBool(IPCDisabledFlag.Name) { + return "" + } + return ctx.GlobalString(IPCPathFlag.Name) +} + // MakeNodeKey creates a node key from set command line flags, either loading it // from a file or as a specified hex value. If neither flags were provided, this // method returns nil and an emphemeral key is to be generated. @@ -593,6 +603,7 @@ func MakeSystemNode(name, version string, extra []byte, ctx *cli.Context) *node. // Configure the node's service container stackConf := &node.Config{ DataDir: MustMakeDataDir(ctx), + IpcPath: MakeIpcPath(ctx), PrivateKey: MakeNodeKey(ctx), Name: MakeNodeName(name, version, ctx), NoDiscovery: ctx.GlobalBool(NoDiscoverFlag.Name), @@ -754,47 +765,6 @@ func MakeChain(ctx *cli.Context) (chain *core.BlockChain, chainDb ethdb.Database return chain, chainDb } -func IpcSocketPath(ctx *cli.Context) (ipcpath string) { - if runtime.GOOS == "windows" { - ipcpath = common.DefaultIpcPath() - if ctx.GlobalIsSet(IPCPathFlag.Name) { - ipcpath = ctx.GlobalString(IPCPathFlag.Name) - } - } else { - ipcpath = common.DefaultIpcPath() - if ctx.GlobalIsSet(DataDirFlag.Name) { - ipcpath = filepath.Join(ctx.GlobalString(DataDirFlag.Name), "geth.ipc") - } - if ctx.GlobalIsSet(IPCPathFlag.Name) { - ipcpath = ctx.GlobalString(IPCPathFlag.Name) - } - } - - return -} - -// StartIPC starts a IPC JSON-RPC API server. -func StartIPC(stack *node.Node, ctx *cli.Context) error { - config := comms.IpcConfig{ - Endpoint: IpcSocketPath(ctx), - } - - initializer := func(conn net.Conn) (comms.Stopper, shared.EthereumApi, error) { - var ethereum *eth.Ethereum - if err := stack.Service(ðereum); err != nil { - return nil, nil, err - } - fe := useragent.NewRemoteFrontend(conn, ethereum.AccountManager()) - xeth := xeth.New(stack, fe) - apis, err := api.ParseApiString(ctx.GlobalString(IPCApiFlag.Name), codec.JSON, xeth, stack) - if err != nil { - return nil, nil, err - } - return xeth, api.Merge(apis...), nil - } - return comms.StartIpc(config, codec.JSON, initializer) -} - // StartRPC starts a HTTP JSON-RPC API server. func StartRPC(stack *node.Node, ctx *cli.Context) error { config := comms.HttpConfig{ diff --git a/common/path.go b/common/path.go index 39eacaceeb..fe8826e0ca 100644 --- a/common/path.go +++ b/common/path.go @@ -87,10 +87,3 @@ func DefaultDataDir() string { // As we cannot guess a stable location, return empty and handle later return "" } - -func DefaultIpcPath() string { - if runtime.GOOS == "windows" { - return `\\.\pipe\geth.ipc` - } - return filepath.Join(DefaultDataDir(), "geth.ipc") -} diff --git a/common/types.go b/common/types.go index acbd5b28da..b1666d7338 100644 --- a/common/types.go +++ b/common/types.go @@ -17,6 +17,8 @@ package common import ( + "encoding/hex" + "encoding/json" "fmt" "math/big" "math/rand" @@ -50,6 +52,21 @@ 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()) +} + // Sets the hash to the value of b. If b is larger than len(h) it will panic func (h *Hash) SetBytes(b []byte) { if len(b) > len(h) { @@ -129,6 +146,38 @@ func (a *Address) Set(other Address) { } } +// Serialize given address to JSON +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/backend.go b/eth/backend.go index 91f02db72a..0f1584aec4 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -43,6 +43,7 @@ import ( "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) const ( @@ -282,6 +283,12 @@ func (s *Ethereum) Protocols() []p2p.Protocol { return s.protocolManager.SubProtocols } +// Apis implements node.Servie, returning all the API handlers the service wants +// to expose. +func (s *Ethereum) Apis() (string, []rpc.Api) { + return "ethereum", []rpc.Api{} +} + // Start implements node.Service, starting all internal goroutines needed by the // Ethereum protocol implementation. func (s *Ethereum) Start(*p2p.Server) error { diff --git a/eth/downloader/downloader_rpc.go b/eth/downloader/downloader_rpc.go new file mode 100644 index 0000000000..fdc47dafbb --- /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..a5cc8000eb --- /dev/null +++ b/eth/filters/filter_rpc.go @@ -0,0 +1,548 @@ +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 returnLogs(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 nil +} + +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 nil +} + +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 nil +} + +func (s *FilterService) GetFilterLogs(filterId string) vm.Logs { + id, ok := s.filterMapping[filterId] + if !ok { + return returnLogs(nil) + } + + if filter := s.filterManager.Get(id); filter != nil { + return returnLogs(filter.Find()) + } + + return returnLogs(nil) +} + +func (s *FilterService) GetFilterChanges(filterId string) interface{} { + s.filterMapMu.Lock() + id, ok := s.filterMapping[filterId] + s.filterMapMu.Unlock() + + if !ok { // filter not found + return []interface{}{} + } + + switch s.getFilterType(id) { + case blockFilterTy: + return returnHashes(s.blockFilterChanged(id)) + case transactionFilterTy: + return returnHashes(s.transactionFilterChanged(id)) + case logFilterTy: + return returnLogs(s.logFilterChanged(id)) + } + + return []interface{}{} +} + +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 +} + +// returnLogs is a helper that will return an empty logs array case the given logs is nil, otherwise is will return the +// given logs. The RPC interfaces defines that always an array is returned. +func returnLogs(logs vm.Logs) vm.Logs { + if logs == nil { + return vm.Logs{} + } + return logs +} + +// returnHashes is a helper that will return an empty hash array case the given hash array is nil, otherwise is will +// return the given hashes. The RPC interfaces defines that always an array is returned. +func returnHashes(hashes []common.Hash) []common.Hash { + if hashes == nil { + return []common.Hash{} + } + return hashes +} \ No newline at end of file diff --git a/eth/web3_rpc.go b/eth/web3_rpc.go new file mode 100644 index 0000000000..a29befe2b6 --- /dev/null +++ b/eth/web3_rpc.go @@ -0,0 +1,23 @@ +package eth + +import ( + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/node" +) + +type Web3Service struct { + stack *node.Node +} + +func NewWeb3Service(stack *node.Node) *Web3Service { + return &Web3Service{stack} +} + +func (s *Web3Service) ClientVersion() string { + return s.stack.Server().Name +} + +func (s *Web3Service) Sha3(data string) string { + return common.ToHex(crypto.Sha3(common.FromHex(data))) +} 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/node/api.go b/node/api.go new file mode 100644 index 0000000000..e21bc7da8e --- /dev/null +++ b/node/api.go @@ -0,0 +1,22 @@ +// Copyright 2015 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 node + +// PublicApi is the collection of public API methods exposed by the node. +type PublicApi struct { + node *Node +} diff --git a/node/config.go b/node/config.go index 93f0ba79d6..bb451a6062 100644 --- a/node/config.go +++ b/node/config.go @@ -23,6 +23,8 @@ import ( "net" "os" "path/filepath" + "runtime" + "strings" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/logger" @@ -49,6 +51,12 @@ type Config struct { // in memory. DataDir string + // IpcPath is the requested location to place the IPC endpoint. If the path is + // a simple file name, it is placed inside the data directory (or on the root + // pipe path on Windows), whereas if it's a resolvable path name (absolute or + // relative), then that specific path is enforced. An empty path disables IPC. + IpcPath string + // This field should be a valid secp256k1 private key that will be used for both // remote peer identification as well as network traffic encryption. If no key // is configured, the preset one is loaded from the data dir, generating it if @@ -90,6 +98,31 @@ type Config struct { MaxPendingPeers int } +// IpcEndpoint resolves an IPC endpoint based on a configured value, taking into +// account the set data folders as well as the designated platform we're currently +// running on. +func (c *Config) IpcEndpoint() string { + // Short circuit if IPC has not been enabled + if c.IpcPath == "" { + return "" + } + // On windows we can only use plain top-level pipes + if runtime.GOOS == "windows" { + if strings.HasPrefix(c.IpcPath, `\\.\pipe\`) { + return c.IpcPath + } + return `\\.\pipe\` + c.IpcPath + } + // Resolve names into the data directory full paths otherwise + if filepath.Base(c.IpcPath) == c.IpcPath { + if c.DataDir == "" { + return filepath.Join(os.TempDir(), c.IpcPath) + } + return filepath.Join(c.DataDir, c.IpcPath) + } + return c.IpcPath +} + // NodeKey retrieves the currently configured private key of the node, checking // first any manually set key, falling back to the one found in the configured // data folder. If no key can be found, a new one is generated. diff --git a/node/config_test.go b/node/config_test.go index f59f3c0fe0..c27c24a8ee 100644 --- a/node/config_test.go +++ b/node/config_test.go @@ -21,6 +21,7 @@ import ( "io/ioutil" "os" "path/filepath" + "runtime" "testing" "github.com/ethereum/go-ethereum/crypto" @@ -60,6 +61,37 @@ func TestDatadirCreation(t *testing.T) { } } +// Tests that IPC paths are correctly resolved to valid endpoints of different +// platforms. +func TestIpcPathResolution(t *testing.T) { + var tests = []struct { + DataDir string + IpcPath string + Windows bool + Endpoint string + }{ + {"", "", false, ""}, + {"data", "", false, ""}, + {"", "geth.ipc", false, "/tmp/geth.ipc"}, + {"data", "geth.ipc", false, "data/geth.ipc"}, + {"data", "./geth.ipc", false, "./geth.ipc"}, + {"data", "/geth.ipc", false, "/geth.ipc"}, + {"", "", true, ``}, + {"data", "", true, ``}, + {"", "geth.ipc", true, `\\.\pipe\geth.ipc`}, + {"data", "geth.ipc", true, `\\.\pipe\geth.ipc`}, + {"data", `\\.\pipe\geth.ipc`, true, `\\.\pipe\geth.ipc`}, + } + for i, test := range tests { + // Only run when platform/test match + if (runtime.GOOS == "windows") == test.Windows { + if endpoint := (&Config{DataDir: test.DataDir, IpcPath: test.IpcPath}).IpcEndpoint(); endpoint != test.Endpoint { + t.Errorf("test %d: IPC endpoint mismatch: have %s, want %s", i, endpoint, test.Endpoint) + } + } + } +} + // Tests that node keys can be correctly created, persisted, loaded and/or made // ephemeral. func TestNodeKeyPersistency(t *testing.T) { diff --git a/node/errors.go b/node/errors.go index bd5ddeb5de..4206974ff9 100644 --- a/node/errors.go +++ b/node/errors.go @@ -32,6 +32,17 @@ func (e *DuplicateServiceError) Error() string { return fmt.Sprintf("duplicate service: %v", e.Kind) } +// DuplicateApiError is returned during Node startup if a registered service +// requests the registration of an already taken API endpoint. +type DuplicateApiError struct { + Namespace string +} + +// Error generates a textual representation of the duplicate API error. +func (e *DuplicateApiError) Error() string { + return fmt.Sprintf("duplicate api endpoint: %s", e.Namespace) +} + // StopError is returned if a Node fails to stop either any of its registered // services or itself. type StopError struct { diff --git a/node/node.go b/node/node.go index 5566bc44bf..99902139d3 100644 --- a/node/node.go +++ b/node/node.go @@ -19,6 +19,8 @@ package node import ( "errors" + "fmt" + "net" "os" "path/filepath" "reflect" @@ -26,7 +28,11 @@ import ( "syscall" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rpc/comms" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) var ( @@ -50,6 +56,9 @@ type Node struct { serviceFuncs []ServiceConstructor // Service constructors (in dependency order) services map[reflect.Type]Service // Currently running services + ipcEndpoint string // IPC endpoint to listen at (empty = IPC disabled) + ipcListener net.Listener // IPC RPC listener socket to serve API requests + stop chan struct{} // Channel to wait for termination notifications lock sync.RWMutex } @@ -85,6 +94,7 @@ func New(conf *Config) (*Node, error) { MaxPendingPeers: conf.MaxPendingPeers, }, serviceFuncs: []ServiceConstructor{}, + ipcEndpoint: conf.IpcEndpoint(), eventmux: new(event.TypeMux), }, nil } @@ -162,6 +172,14 @@ func (n *Node) Start() error { // Mark the service started for potential cleanup started = append(started, kind) } + // Lastly start the configured RPC interfaces + if err := n.startRpc(services); err != nil { + for _, service := range services { + service.Stop() + } + running.Stop() + return err + } // Finish initializing the startup n.services = services n.server = running @@ -170,6 +188,78 @@ func (n *Node) Start() error { return nil } +// startRpc initializes and starts the IPC and HTTP RPC endpoints. +func (n *Node) startRpc(services map[reflect.Type]Service) error { + // Gather and register all the APIs exposed by the services + ipcHandler := rpc.NewServer() + httpHandler := rpc.NewServer() + + taken := make(map[string]struct{}) + for _, service := range services { + name, apis := service.Apis() + for _, api := range apis { + // Ensure each API and version is unique + namespace := fmt.Sprintf("%s.v%d", name, api.Version) + if _, ok := taken[namespace]; ok { + return &DuplicateApiError{Namespace: namespace} + } + taken[namespace] = struct{}{} + + // Namespace was free, register all the method handlers + for _, handler := range api.Handlers { + // Resolve the registration path + path := namespace + if handler.Path != "" { + path += "." + handler.Path + } + // Inject the methods into all API servers + if err := ipcHandler.RegisterName(path, handler.Handler); err != nil { + return err + } + if err := httpHandler.RegisterName(path, handler.Handler); err != nil { + return err + } + } + } + } + // All APIs registered, start the IPC and HTTP listeners + var ( + ipcListener net.Listener + err error + ) + if n.ipcEndpoint != "" { + if ipcListener, err = comms.CreateListener(comms.IpcConfig{Endpoint: n.ipcEndpoint}); err != nil { + return err + } + go func() { + glog.V(logger.Info).Infof("IPC endpoint opened: %s", n.ipcEndpoint) + defer glog.V(logger.Info).Infof("IPC endpoint closed: %s", n.ipcEndpoint) + + for { + conn, err := ipcListener.Accept() + if err != nil { + // Terminate if the listener was closed + n.lock.RLock() + closed := n.ipcListener == nil + n.lock.RUnlock() + if closed { + return + } + // Not closed, just some error; report and continue + glog.V(logger.Error).Infof("IPC accept failed: %v", err) + continue + } + codec := rpc.NewJSONCodec(conn) + go ipcHandler.ServeCodec(codec) + } + }() + } + // All listeners booted successfully + n.ipcListener = ipcListener + + return nil +} + // Stop terminates a running node along with all it's services. In the node was // not started, an error is returned. func (n *Node) Stop() error { @@ -180,7 +270,11 @@ func (n *Node) Stop() error { if n.server == nil { return ErrNodeStopped } - // Otherwise terminate all the services and the P2P server too + // Otherwise terminate the API, all services and the P2P server too + if n.ipcListener != nil { + n.ipcListener.Close() + n.ipcListener = nil + } failure := &StopError{ Services: make(map[reflect.Type]error), } diff --git a/node/node_example_test.go b/node/node_example_test.go index 2f9b49a56d..1af191f207 100644 --- a/node/node_example_test.go +++ b/node/node_example_test.go @@ -23,6 +23,7 @@ import ( "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) // SampleService is a trivial network service that can be attached to a node for @@ -30,11 +31,13 @@ import ( // // The following methods are needed to implement a node.Service: // - Protocols() []p2p.Protocol - devp2p protocols the service can communicate on +// - Apis() (string, []rpc.Api) - api methods the service wants to expose on rpc channels // - Start() error - method invoked when the node is ready to start the service // - Stop() error - method invoked when the node terminates the service type SampleService struct{} func (s *SampleService) Protocols() []p2p.Protocol { return nil } +func (s *SampleService) Apis() (string, []rpc.Api) { return "", nil } func (s *SampleService) Start(*p2p.Server) error { fmt.Println("Service starting..."); return nil } func (s *SampleService) Stop() error { fmt.Println("Service stopping..."); return nil } diff --git a/node/node_test.go b/node/node_test.go index ef096fc917..41f27ace22 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -18,19 +18,27 @@ package node import ( "errors" + "fmt" "io/ioutil" + "math/rand" "os" "reflect" "testing" + "time" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rpc/codec" + "github.com/ethereum/go-ethereum/rpc/comms" + "github.com/ethereum/go-ethereum/rpc/shared" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) var ( testNodeKey, _ = crypto.GenerateKey() testNodeConfig = &Config{ + IpcPath: fmt.Sprintf("test-%d.ipc", rand.Int63()), PrivateKey: testNodeKey, Name: "test node", } @@ -494,3 +502,81 @@ func TestProtocolGather(t *testing.T) { } } } + +// Tests that all APIs defined by individual services get exposed. +func TestApiGather(t *testing.T) { + stack, err := New(testNodeConfig) + if err != nil { + t.Fatalf("failed to create protocol stack: %v", err) + } + // Register a batch of services with some configured APIs + calls := make(chan string, 1) + + services := map[string]struct { + Namespace string + Apis []rpc.Api + Maker InstrumentingWrapper + }{ + "Zero APIs": {"", []rpc.Api{}, InstrumentedServiceMakerA}, + "Single API": {"single", []rpc.Api{ + {1, []rpc.ApiHandler{{Handler: &OneMethodApi{fun: func() { calls <- "single.v1" }}}}}, + }, InstrumentedServiceMakerB}, + "Many APIs": {"multi", []rpc.Api{ + {1, []rpc.ApiHandler{{Handler: &OneMethodApi{fun: func() { calls <- "multi.v1" }}}}}, + {2, []rpc.ApiHandler{ + {Handler: &OneMethodApi{fun: func() { calls <- "multi.v2" }}}, + {Path: "nested", Handler: &OneMethodApi{fun: func() { calls <- "multi.v2.nested" }}}, + }}, + }, InstrumentedServiceMakerC}, + } + for id, config := range services { + config := config + constructor := func(*ServiceContext) (Service, error) { + return &InstrumentedService{ + apiNamespace: config.Namespace, + apiHandlers: config.Apis, + }, nil + } + if err := stack.Register(config.Maker(constructor)); err != nil { + t.Fatalf("service %s: registration failed: %v", id, err) + } + } + // Start the services and ensure all API start successfully + if err := stack.Start(); err != nil { + t.Fatalf("failed to start protocol stack: %v", err) + } + defer stack.Stop() + + // Connect to the RPC server and verify the various registered endpoints + ipcClient, err := comms.NewIpcClient(comms.IpcConfig{Endpoint: testNodeConfig.IpcEndpoint()}, codec.JSON) + if err != nil { + t.Fatalf("failed to connect to the IPC API server: %v", err) + } + + tests := []struct { + Method string + Result string + }{ + {"single.v1.theOneMethod", "single.v1"}, + {"multi.v1.theOneMethod", "multi.v1"}, + {"multi.v2.theOneMethod", "multi.v2"}, + {"multi.v2.nested.theOneMethod", "multi.v2.nested"}, + } + for i, test := range tests { + if err := ipcClient.Send(shared.Request{Id: i, Jsonrpc: "2.0", Method: test.Method}); err != nil { + t.Fatalf("test %d: failed to send API request: %v", i, err) + } + _, err := ipcClient.Recv() + if err != nil { + t.Fatalf("test %d: failed to read API reply: %v", i, err) + } + select { + case result := <-calls: + if result != test.Result { + t.Errorf("test %d: result mismatch: have %s, want %s", i, result, test.Result) + } + case <-time.After(time.Second): + t.Fatalf("test %d: rpc execution timeout", i) + } + } +} diff --git a/node/service.go b/node/service.go index bfeeb7ab98..200e8897f3 100644 --- a/node/service.go +++ b/node/service.go @@ -23,6 +23,7 @@ import ( "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/p2p" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) // ServiceContext is a collection of service independent options inherited from @@ -70,6 +71,10 @@ type Service interface { // Protocol retrieves the P2P protocols the service wishes to start. Protocols() []p2p.Protocol + // Apis retrieves the list of versioned APIs the service provides to register + // under the requested top level API namespace. + Apis() (string, []rpc.Api) + // Start is called after all services have been constructed and the networking // layer was also initialized to spawn any goroutines required by the service. Start(server *p2p.Server) error diff --git a/node/utils_test.go b/node/utils_test.go index 756622c863..0491fc7c38 100644 --- a/node/utils_test.go +++ b/node/utils_test.go @@ -23,12 +23,14 @@ import ( "reflect" "github.com/ethereum/go-ethereum/p2p" + rpc "github.com/ethereum/go-ethereum/rpc/v2" ) // NoopService is a trivial implementation of the Service interface. type NoopService struct{} func (s *NoopService) Protocols() []p2p.Protocol { return nil } +func (s *NoopService) Apis() (string, []rpc.Api) { return "", nil } func (s *NoopService) Start(*p2p.Server) error { return nil } func (s *NoopService) Stop() error { return nil } @@ -49,11 +51,14 @@ func NewNoopServiceD(*ServiceContext) (Service, error) { return new(NoopServiceD // InstrumentedService is an implementation of Service for which all interface // methods can be instrumented both return value as well as event hook wise. type InstrumentedService struct { - protocols []p2p.Protocol - start error - stop error + protocols []p2p.Protocol + apiNamespace string + apiHandlers []rpc.Api + start error + stop error protocolsHook func() + apisHook func() startHook func(*p2p.Server) stopHook func() } @@ -67,6 +72,13 @@ func (s *InstrumentedService) Protocols() []p2p.Protocol { return s.protocols } +func (s *InstrumentedService) Apis() (string, []rpc.Api) { + if s.apisHook != nil { + s.apisHook() + } + return s.apiNamespace, s.apiHandlers +} + func (s *InstrumentedService) Start(server *p2p.Server) error { if s.startHook != nil { s.startHook(server) @@ -115,3 +127,14 @@ func InstrumentedServiceMakerB(base ServiceConstructor) ServiceConstructor { func InstrumentedServiceMakerC(base ServiceConstructor) ServiceConstructor { return InstrumentingWrapperMaker(base, reflect.TypeOf(InstrumentedServiceC{})) } + +// OneMethodApi is a single-method API handler to be returned by test services. +type OneMethodApi struct { + fun func() +} + +func (api *OneMethodApi) TheOneMethod() { + if api.fun != nil { + api.fun() + } +} 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 d6820cd2ef..794b6abeea 100644 --- a/rpc/api/utils.go +++ b/rpc/api/utils.go @@ -191,6 +191,8 @@ func ParseApiString(apistr string, codec codec.Codec, xeth *xeth.XEth, stack *no 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..16d735c687 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 @@ -108,6 +124,11 @@ func StartIpc(cfg IpcConfig, codec codec.Codec, initializer InitFunc) error { return nil } +// CreateListener creates an listener, on Unix platforms this is a unix socket, on Windows this is a named pipe +func CreateListener(cfg IpcConfig) (net.Listener, error) { + return ipcListen(cfg) +} + func ipcLoop(cfg IpcConfig, codec codec.Codec, initializer InitFunc, l net.Listener) { glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint) defer os.Remove(cfg.Endpoint) diff --git a/rpc/v2/api.go b/rpc/v2/api.go new file mode 100644 index 0000000000..477ae50745 --- /dev/null +++ b/rpc/v2/api.go @@ -0,0 +1,30 @@ +// Copyright 2015 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 v2 + +// Api describes a versioned API schema offered over the RPC interface. +type Api struct { + Version int // Version number under which to advertise this method set + Handlers []ApiHandler // List of API handlers advertized on this API version +} + +// ApiHandler describes the set of methods offered over the RPC interface +type ApiHandler struct { + Path string // Path under which the RPC methods of Handler are to be exposed + Handler interface{} // Receiver instance which holds the methods to expose + Public bool // Indication if the methods can be considered safe for public use +} diff --git a/rpc/v2/doc.go b/rpc/v2/doc.go new file mode 100644 index 0000000000..e51494adb0 --- /dev/null +++ b/rpc/v2/doc.go @@ -0,0 +1,102 @@ +// Copyright 2015 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 rpc provides access to the exported methods of an object across a network +or other I/O connection. After creating a server instance objects can be registered, +making it visible from the outside. Exported methods that follow specific +conventions can be called remotely. It also has support for the publish/subscribe +pattern. + +Methods that satisfy the following criteria are made available for remote access: + - object must be exported + - method must be exported + - method returns 0, 1 (response or error) or 2 (response and error) values + - method argument(s) must be exported or builtin types + - method returned value(s) must be exported or builtin types + +An example method: + func (s *CalcService) Div(a, b int) (int, error) + +When the returned error isn't nil the returned integer is ignored and the error is +send back to the client. Otherwise the returned integer is send back to the client. + +The server offers the ServeCodec method which accepts a ServerCodec instance. It will +read requests from the codec, process the request and sends the response back to the +client using the codec. The server can execute requests concurrently. Responses +can be send back to the client out of order. + +An example server which uses the JSON codec: + type CalculatorService struct {} + + func (s *CalculatorService) Add(a, b int) int { + return a + b + } + + func (s *CalculatorService Div(a, b int) (int, error) { + if b == 0 { + return 0, errors.New("divide by zero") + } + return a/b, nil + } + + calculator := new(CalculatorService) + server := NewServer() + server.RegisterName("calculator", calculator") + + l, _ := net.ListenUnix("unix", &net.UnixAddr{Net: "unix", Name: "/tmp/calculator.sock"}) + for { + c, _ := l.AcceptUnix() + codec := v2.NewJSONCodec(c) + go server.ServeCodec(codec) + } + +The package also supports the publish subscribe pattern through the use of subscriptions. +A method that is considered eligible for notifications must satisfy the following criteria: + - object must be exported + - method must be exported + - method argument(s) must be exported or builtin types + - method must return the tuple Subscription, error + + +An example method: + func (s *BlockChainService) Head() (Subscription, error) { + sub := s.bc.eventMux.Subscribe(ChainHeadEvent{}) + return v2.NewSubscription(sub), nil + } + +This method will push all raised ChainHeadEvents to subscribed clients. If the client is only +interested in every N'th block it is possible to add a criteria. + + func (s *BlockChainService) HeadFiltered(nth uint64) (Subscription, error) { + sub := s.bc.eventMux.Subscribe(ChainHeadEvent{}) + + criteria := func(event interface{}) bool { + chainHeadEvent := event.(ChainHeadEvent) + if chainHeadEvent.Block.NumberU64() % nth == 0 { + return true + } + return false + } + + return v2.NewSubscriptionFiltered(sub, criteria), nil + } + +Subscriptions are deleted when: + - the user sends an unsubscribe request + - the connection which was used to create the subscription is closed +*/ +package v2 diff --git a/rpc/v2/errors.go b/rpc/v2/errors.go new file mode 100644 index 0000000000..a06d19d84c --- /dev/null +++ b/rpc/v2/errors.go @@ -0,0 +1,85 @@ +// Copyright 2015 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 v2 + +import "fmt" + +// request is for an unknown service +type methodNotFoundError struct { + service string + method string +} + +func (e *methodNotFoundError) Code() int { + return -32601 +} + +func (e *methodNotFoundError) Error() string { + return fmt.Sprintf("The method %s%s%s does not exist/is not available", e.service, serviceMethodSeparator, e.method) +} + +// received message isn't a valid request +type invalidRequestError struct { + message string +} + +func (e *invalidRequestError) Code() int { + return -32600 +} + +func (e *invalidRequestError) Error() string { + return e.message +} + +// received message is invalid +type invalidMessageError struct { + message string +} + +func (e *invalidMessageError) Code() int { + return -32700 +} + +func (e *invalidMessageError) Error() string { + return e.message +} + +// unable to decode supplied params, or an invalid number of parameters +type invalidParamsError struct { + message string +} + +func (e *invalidParamsError) Code() int { + return -32602 +} + +func (e *invalidParamsError) Error() string { + return e.message +} + +// logic error, callback returned an error +type callbackError struct { + message string +} + +func (e *callbackError) Code() int { + return -32000 +} + +func (e *callbackError) Error() string { + return e.message +} diff --git a/rpc/v2/json.go b/rpc/v2/json.go new file mode 100644 index 0000000000..54f2899e26 --- /dev/null +++ b/rpc/v2/json.go @@ -0,0 +1,336 @@ +// Copyright 2015 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 v2 + +import ( + "encoding/json" + "fmt" + "io" + "reflect" + "strings" + "sync/atomic" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" +) + +const ( + jsonRPCVersion = "2.0" + serviceMethodSeparator = "_" + subscribeMethod = "eth_subscribe" + unsubscribeMethod = "eth_unsubscribe" + notificationMethod = "eth_subscription" +) + +// JSON-RPC request +type jsonRequest struct { + Method string `json:"method"` + Version string `json:"jsonrpc"` + Id int64 `json:"id"` + Payload json.RawMessage `json:"params"` +} + +// JSON-RPC response +type jsonSuccessResponse struct { + Version string `json:"jsonrpc"` + Id int64 `json:"id"` + Result interface{} `json:"result,omitempty"` +} + +// JSON-RPC error object +type jsonError struct { + Code int `json:"code"` + Message string `json:"message"` + Data interface{} `json:"data,omitempty"` +} + +// JSON-RPC error response +type jsonErrResponse struct { + Version string `json:"jsonrpc"` + Id *int64 `json:"id,omitempty"` + Error jsonError `json:"error"` +} + +// JSON-RPC notification payload +type jsonSubscription struct { + Subscription string `json:"subscription"` + Result interface{} `json:"result,omitempty"` +} + +// JSON-RPC notification +type jsonNotification struct { + Version string `json:"jsonrpc"` + Method string `json:"method"` + Params jsonSubscription `json:"params"` +} + +// jsonCodec reads and writes JSON-RPC messages to the underlying connection. It also has support for parsing arguments +// and serializing (result) objects. +type jsonCodec struct { + closed chan interface{} + isClosed int32 + d *json.Decoder + e *json.Encoder + req jsonRequest + rw io.ReadWriteCloser +} + +// NewJSONCodec creates a new RPC server codec with support for JSON-RPC 2.0 +func NewJSONCodec(rwc io.ReadWriteCloser) ServerCodec { + d := json.NewDecoder(rwc) + d.UseNumber() + return &jsonCodec{closed: make(chan interface{}), d: d, e: json.NewEncoder(rwc), rw: rwc, isClosed: 0} +} + +// isBatch returns true when the first non-whitespace characters is '[' +func isBatch(msg json.RawMessage) bool { + for _, c := range msg { + // skip insignificant whitespace (http://www.ietf.org/rfc/rfc4627.txt) + if c == 0x20 || c == 0x09 || c == 0x0a || c == 0x0d { + continue + } + return c == '[' + } + return false +} + +// ReadRequestHeaders will read new requests without parsing the arguments. It will return a collection of requests, an +// indication if these requests are in batch form or an error when the incoming message could not be read/parsed. +func (c *jsonCodec) ReadRequestHeaders() ([]rpcRequest, bool, RPCError) { + var incomingMsg json.RawMessage + if err := c.d.Decode(&incomingMsg); err != nil { + return nil, false, &invalidRequestError{err.Error()} + } + + if isBatch(incomingMsg) { + return parseBatchRequest(incomingMsg) + } + + return parseRequest(incomingMsg) +} + +// parseRequest will parse a single request from the given RawMessage. It will return the parsed request, an indication +// if the request was a batch or an error when the request could not be parsed. +func parseRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, RPCError) { + var in jsonRequest + if err := json.Unmarshal(incomingMsg, &in); err != nil { + return nil, false, &invalidMessageError{err.Error()} + } + + // subscribe are special, they will always use `subscribeMethod` as service method + if in.Method == subscribeMethod { + reqs := []rpcRequest{rpcRequest{id: in.Id, isPubSub: true}} + if len(in.Payload) > 0 { + // 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"} + } + + if in.Method == unsubscribeMethod { + return []rpcRequest{rpcRequest{id: in.Id, isPubSub: true, + method: unsubscribeMethod, params: in.Payload}}, false, nil + } + + // regular RPC call + elems := strings.Split(in.Method, serviceMethodSeparator) + if len(elems) != 2 { + return nil, false, &methodNotFoundError{in.Method, ""} + } + + if len(in.Payload) == 0 { + return []rpcRequest{rpcRequest{service: elems[0], method: elems[1], id: in.Id}}, false, nil + } + + return []rpcRequest{rpcRequest{service: elems[0], method: elems[1], id: in.Id, params: in.Payload}}, false, nil +} + +// parseBatchRequest will parse a batch request into a collection of requests from the given RawMessage, an indication +// if the request was a batch or an error when the request could not be read. +func parseBatchRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, RPCError) { + var in []jsonRequest + if err := json.Unmarshal(incomingMsg, &in); err != nil { + return nil, false, &invalidMessageError{err.Error()} + } + + requests := make([]rpcRequest, len(in)) + for i, r := range in { + // (un)subscribe are special, they will always use the same service.method + if r.Method == subscribeMethod { + requests[i] = rpcRequest{id: r.Id, isPubSub: true} + if len(r.Payload) > 0 { + 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"} + } + + if r.Method == unsubscribeMethod { + requests[i] = rpcRequest{id: r.Id, isPubSub: true, method: unsubscribeMethod, params: r.Payload} + continue + } + + elems := strings.Split(r.Method, serviceMethodSeparator) + if len(elems) != 2 { + return nil, true, &methodNotFoundError{r.Method, ""} + } + + if len(r.Payload) == 0 { + requests[i] = rpcRequest{service: elems[0], method: elems[1], id: r.Id, params: nil} + } else { + requests[i] = rpcRequest{service: elems[0], method: elems[1], id: r.Id, params: r.Payload} + } + } + + return requests, true, nil +} + +// ParseRequestArguments tries to parse the given params (json.RawMessage) with the given types. It returns the parsed +// values or an error when the parsing failed. +func (c *jsonCodec) ParseRequestArguments(argTypes []reflect.Type, params interface{}) ([]reflect.Value, RPCError) { + if args, ok := params.(json.RawMessage); !ok { + return nil, &invalidParamsError{"Invalid params supplied"} + } else { + return parsePositionalArguments(args, argTypes) + } +} + +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) + } + } + } + + if err := json.Unmarshal(args, ¶ms); err != nil { + return nil, &invalidParamsError{err.Error()} + } + + // Convert pointers back to values where necessary + for i, a := range argValues { + if a.Kind() != argTypes[i].Kind() { + argValues[i] = reflect.Indirect(argValues[i]) + } + } + + return argValues, nil +} + +// 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} +} + +// CreateErrorResponse will create a JSON-RPC error response with the given id and error. +func (c *jsonCodec) CreateErrorResponse(id *int64, err RPCError) interface{} { + return &jsonErrResponse{Version: jsonRPCVersion, Id: id, Error: jsonError{Code: err.Code(), Message: err.Error()}} +} + +// CreateErrorResponseWithInfo will create a JSON-RPC error response with the given id and error. +// info is optional and contains additional information about the error. When an empty string is passed it is ignored. +func (c *jsonCodec) CreateErrorResponseWithInfo(id *int64, err RPCError, info interface{}) interface{} { + return &jsonErrResponse{Version: jsonRPCVersion, Id: id, + Error: jsonError{Code: err.Code(), Message: err.Error(), Data: info}} +} + +// 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}} +} + +// Write message to client +func (c *jsonCodec) Write(res interface{}) error { + return c.e.Encode(res) +} + +// Close the underlying connection +func (c *jsonCodec) Close() { + if atomic.CompareAndSwapInt32(&c.isClosed, 0, 1) { + close(c.closed) + c.rw.Close() + } +} + +// Closed returns a channel which will be closed when Close is called +func (c *jsonCodec) Closed() <-chan interface{} { + return c.closed +} diff --git a/rpc/v2/json_test.go b/rpc/v2/json_test.go new file mode 100644 index 0000000000..dc8a345d7f --- /dev/null +++ b/rpc/v2/json_test.go @@ -0,0 +1,73 @@ +package v2 + +import ( + "bufio" + "bytes" + "reflect" + "testing" +) + +type RWC struct { + *bufio.ReadWriter +} + +func (rwc *RWC) Close() error { + return nil +} + +func TestJSONRequestParsing(t *testing.T) { + server := NewServer() + service := new(Service) + + if err := server.RegisterName("calc", service); err != nil { + t.Fatalf("%v", err) + } + + req := bytes.NewBufferString(`{"id": 1234, "jsonrpc": "2.0", "method": "calc_add", "params": [11, 22]}`) + var str string + reply := bytes.NewBufferString(str) + rw := &RWC{bufio.NewReadWriter(bufio.NewReader(req), bufio.NewWriter(reply))} + + codec := NewJSONCodec(rw) + + requests, batch, err := codec.ReadRequestHeaders() + if err != nil { + t.Fatalf("%v", err) + } + + if batch { + t.Fatalf("Request isn't a batch") + } + + if len(requests) != 1 { + t.Fatalf("Expected 1 request but got %d requests - %v", len(requests), requests) + } + + if requests[0].service != "calc" { + t.Fatalf("Expected service 'calc' but got '%s'", requests[0].service) + } + + if requests[0].method != "add" { + t.Fatalf("Expected method 'Add' but got '%s'", requests[0].method) + } + + if requests[0].id != 1234 { + t.Fatalf("Expected id 1234 but got %d", requests[0].id) + } + + var arg int + args := []reflect.Type{reflect.TypeOf(arg), reflect.TypeOf(arg)} + + v, err := codec.ParseRequestArguments(args, requests[0].params) + if err != nil { + t.Fatalf("%v", err) + } + + if len(v) != 2 { + t.Fatalf("Expected 2 argument values, got %d", len(v)) + } + + if v[0].Int() != 11 || v[1].Int() != 22 { + t.Fatalf("expected %d == 11 && %d == 22", v[0].Int(), v[1].Int()) + } +} diff --git a/rpc/v2/server.go b/rpc/v2/server.go new file mode 100644 index 0000000000..2fa434d4c1 --- /dev/null +++ b/rpc/v2/server.go @@ -0,0 +1,480 @@ +// Copyright 2015 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 v2 + +import ( + "fmt" + "reflect" + + "runtime" + + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" +) + +// NewServer will create a new server instance with no registered handlers. +func NewServer() *Server { + server := &Server{services: make(serviceRegistry), subscriptions: make(subscriptionRegistry)} + + // register a default service which will provide meta information about the RPC service such as the services and + // methods it offers. + 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 +// with a dot between the rcrv and method, e.g. Calculator.Add. The supplied +// rcvr must be a pointer. +func (s *Server) Register(rcvr interface{}) error { + return s.register(rcvr, "", false) +} + +// RegisterName publishes suitable methods of rcvr. Methods will be published +// with a dot between the given name and method, e.g. Calculator.Add. The +// supplied rcvr must be a pointer. +func (s *Server) RegisterName(name string, rcvr interface{}) error { + return s.register(rcvr, name, true) +} + +// register will create an service for the given rcvr type under the given name if specified in useName. +// When no methods on the given rcvr match the criteria to be either a RPC method or a subscription an error is +// returned. Otherwise a new service is created and added to the service collection this server instance serves. +func (s *Server) register(rcvr interface{}, name string, useName bool) error { + if s.services == nil { + s.services = make(serviceRegistry) + } + + svc := new(service) + svc.typ = reflect.TypeOf(rcvr) + rcvrVal := reflect.ValueOf(rcvr) + + sname := reflect.Indirect(rcvrVal).Type().Name() + if useName { + sname = name + } + if sname == "" { + return fmt.Errorf("no service name for type %s", svc.typ.String()) + } + if !isExported(sname) && !useName { + return fmt.Errorf("%s is not exported", sname) + } + if !useName { + sname = formatName(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(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") + } + + s.services[svc.name] = svc + + return nil +} + +// ServeCodec reads incoming requests from codec, calls the appropriate callback and writes the +// response back using the given codec. It will block until the codec is closed. +// +// This server will: +// 1. allow for asynchronous and parallel request execution +// 2. supports notifications (pub/sub) +// 3. supports request batches +func (s *Server) ServeCodec(codec ServerCodec) { + defer func() { + if err := recover(); err != nil { + const size = 64 << 10 + buf := make([]byte, size) + buf = buf[:runtime.Stack(buf, false)] + glog.Errorln(string(buf)) + } + codec.Close() + }() + + for { + reqs, batch, err := s.readRequest(codec) + if err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + codec.Write(codec.CreateErrorResponse(nil, err)) + break + } + + if batch { + go s.execBatch(codec, reqs) + } else { + go s.exec(codec, reqs[0]) + } + } +} + +// sendNotification will create a notification from the given event by serializing member fields of the event. +// 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{}) { + notification := codec.CreateNotification(subid, event) + + if err := codec.Write(notification); err != nil { + codec.Close() + } +} + +// createSubscription will register a new subscription and waits for raised events. When an event is raised it will: +// 1. test if the event is raised matches the criteria the user has (optionally) specified +// 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.callb.rcvr} + if len(req.args) > 0 { + args = append(args, req.args...) + } + + subid, err := newSubscriptionId() + if err != nil { + return "", err + } + + reply := req.callb.method.Func.Call(args) + + if reply[1].IsNil() { // no error + if subscription, ok := reply[0].Interface().(Subscription); ok { + s.muSubcriptions.Lock() + s.subscriptions[subid] = subscription + s.muSubcriptions.Unlock() + go func() { + cases := []reflect.SelectCase{ + reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(subscription.Chan())}, // new event + reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(c.Closed())}, // connection closed + } + + for { + idx, notification, recvOk := reflect.Select(cases) + switch idx { + case 0: // new event, or channel closed + if recvOk { // send notification + if event, ok := notification.Interface().(*event.Event); ok { + if subscription.match == nil || subscription.match(event.Data) { + sendNotification(c, subid, subscription.format(event.Data)) + } + } + } else { // user send an eth_unsubscribe request + return + } + case 1: // connection closed + s.unsubscribe(subid) + return + } + } + }() + } else { // unable to create subscription + s.muSubcriptions.Lock() + delete(s.subscriptions, subid) + s.muSubcriptions.Unlock() + } + } else { + return "", fmt.Errorf("Unable to create subscription") + } + + return subid, nil +} + +// unsubscribe calls the Unsubscribe method on the subscription and removes a subscription from the subscription +// registry. +func (s *Server) unsubscribe(subid string) bool { + s.muSubcriptions.Lock() + defer s.muSubcriptions.Unlock() + if sub, ok := s.subscriptions[subid]; ok { + sub.Unsubscribe() + delete(s.subscriptions, subid) + return true + } + return false +} + +// exec executes the given request and writes the result back using the codec. +func (s *Server) exec(codec ServerCodec, req *serverRequest) { + if req.err != nil { // error during request parsing + rpcErr := codec.CreateErrorResponse(&req.id, req.err) + if err := codec.Write(rpcErr); err != nil { + codec.Close() + } + return + } + + if req.isUnsubscribe { // first param must be the subscription id + if len(req.args) >= 1 && req.args[0].Kind() == reflect.String { + subid := req.args[0].String() + if s.unsubscribe(subid) { + if err := codec.Write(codec.CreateResponse(req.id, true)); err != nil { + codec.Close() + } + } else { + rpcErr := codec.CreateErrorResponse(&req.id, + &callbackError{fmt.Sprintf("subscription '%s' not found", subid)}) + if err := codec.Write(rpcErr); err != nil { + codec.Close() + } + } + } else { + rpcErr := codec.CreateErrorResponse(&req.id, &invalidParamsError{"Expected subscription id as argument"}) + if err := codec.Write(rpcErr); err != nil { + codec.Close() + } + } + return + } + + if req.callb.isSubscribe { + subid, err := s.createSubscription(codec, req) + var response interface{} + if err == nil { + response = codec.CreateResponse(req.id, subid) + } else { + response = codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}) + } + + if err = codec.Write(response); err != nil { + codec.Close() + } + return + } + + // regular RPC call + if len(req.args) != len(req.callb.argTypes) { + rpcErr := &invalidParamsError{fmt.Sprintf("%s%s%s expects %d parameters, got %d", + req.svcname, serviceMethodSeparator, req.callb.method.Name, + len(req.callb.argTypes), len(req.args))} + + res := codec.CreateErrorResponse(&req.id, rpcErr) + if err := codec.Write(res); err != nil { + codec.Close() + } + return + } + + arguments := []reflect.Value{req.callb.rcvr} + if len(req.args) > 0 { + arguments = append(arguments, req.args...) + } + + reply := req.callb.method.Func.Call(arguments) + + if len(reply) == 0 { + if err := codec.Write(codec.CreateResponse(req.id, nil)); err != nil { + codec.Close() + } + return + } + + if req.callb.errPos >= 0 { // test if method returned an error + if !reply[req.callb.errPos].IsNil() { + e := reply[req.callb.errPos].Interface().(error) + res := codec.CreateErrorResponse(&req.id, &callbackError{e.Error()}) + if err := codec.Write(res); err != nil { + codec.Close() + } + return + } + } + + if err := codec.Write(codec.CreateResponse(req.id, reply[0].Interface())); err != nil { + codec.Close() + } +} + +// execBatch executes the given requests and writes the result back using the codec. It will only write the response +// back when the last request is processed. +func (s *Server) execBatch(codec ServerCodec, requests []*serverRequest) { + responses := make([]interface{}, len(requests)) + + for i, req := range requests { + if req.err != nil { // error during parsing of request + responses[i] = codec.CreateErrorResponse(&req.id, req.err) + continue + } + + if req.isUnsubscribe { + if len(req.args) == 1 && req.args[0].Kind() == reflect.String { + subid := req.args[0].String() + if s.unsubscribe(subid) { + responses[i] = codec.CreateResponse(req.id, true) + } else { + e := &callbackError{fmt.Sprintf("subscription '%s' not found", subid)} + responses[i] = codec.CreateErrorResponse(&req.id, e) + } + } else { + e := &invalidParamsError{"Expected subscription id as argument"} + responses[i] = codec.CreateErrorResponse(&req.id, e) + } + continue + } + + if req.callb.isSubscribe { + subid, err := s.createSubscription(codec, req) + var response interface{} + if err == nil { + response = codec.CreateResponse(req.id, subid) + } else { + response = codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}) + } + + responses[i] = response + continue + } + + var reply []reflect.Value + + if len(req.args) != len(req.callb.argTypes) { + rpcErr := &invalidParamsError{fmt.Sprintf("%s%s%s expects %d parameters, got %d", + req.svcname, serviceMethodSeparator, req.callb.method.Name, len(req.callb.argTypes), len(req.args))} + responses[i] = codec.CreateErrorResponse(&req.id, rpcErr) + continue + } + + arguments := []reflect.Value{req.callb.rcvr} + if len(req.args) > 0 { + arguments = append(arguments, req.args...) + } + + reply = req.callb.method.Func.Call(arguments) + + if len(reply) == 0 { + responses[i] = codec.CreateResponse(req.id, nil) + continue + } + + if req.callb.errPos >= 0 { + if !reply[req.callb.errPos].IsNil() { + if e, ok := reply[req.callb.errPos].Interface().(error); ok { + rpcErr := &callbackError{e.Error()} + responses[i] = codec.CreateErrorResponse(&req.id, rpcErr) + continue + } + } + } + + responses[i] = codec.CreateResponse(req.id, reply[0].Interface()) + } + + if err := codec.Write(responses); err != nil { + glog.V(logger.Error).Infof("%v\n", err) + codec.Close() + } +} + +// readRequest requests the next (batch) request from the codec. It will return the collection of requests, an +// indication if the request was a batch, the invalid request identifier and an error when the request could not be +// read/parsed. +func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, RPCError) { + reqs, batch, err := codec.ReadRequestHeaders() + if err != nil { + return nil, batch, err + } + + requests := make([]*serverRequest, len(reqs)) + + // verify requests + for i, r := range reqs { + var ok bool + var svc *service + + if r.isPubSub && r.method == unsubscribeMethod { + requests[i] = &serverRequest{id: r.id, isUnsubscribe: true} + argTypes := []reflect.Type{reflect.TypeOf("")} + if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil { + requests[i].args = args + } else { + requests[i].err = &invalidParamsError{err.Error()} + } + continue + } + + if svc, ok = s.services[r.service]; !ok { + requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}} + continue + } + + if r.isPubSub { // eth_subscribe + if callb, ok := svc.subscriptions[r.method]; ok { + 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...) + if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil { + requests[i].args = args[1:] // first one is service.method name which isn't an actual argument + } else { + requests[i].err = &invalidParamsError{err.Error()} + } + } + } else { + requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{subscribeMethod, r.method}} + } + continue + } + + if callb, ok := svc.callbacks[r.method]; ok { + 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 + } else { + requests[i].err = &invalidParamsError{err.Error()} + } + } + continue + } + + requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}} + } + + return requests, batch, nil +} diff --git a/rpc/v2/server_test.go b/rpc/v2/server_test.go new file mode 100644 index 0000000000..ea08c6ccc9 --- /dev/null +++ b/rpc/v2/server_test.go @@ -0,0 +1,244 @@ +package v2 + +import ( + "encoding/json" + "fmt" + "reflect" + "testing" + "time" +) + +type Service struct{} + +type Args struct { + S string +} + +func (s *Service) NoArgsRets() { +} + +type Result struct { + String string + Int int + Args *Args +} + +func (s *Service) Echo(str string, i int, args *Args) Result { + return Result{str, i, args} +} + +func (s *Service) Rets() (string, error) { + return "", nil +} + +func (s *Service) InvalidRets1() (error, string) { + return nil, "" +} + +func (s *Service) InvalidRets2() (string, string) { + return "", "" +} + +func (s *Service) InvalidRets3() (string, string, error) { + return "", "", nil +} + +func (s *Service) Subscription() (Subscription, error) { + return NewSubscription(nil), nil +} + +func TestServerRegister(t *testing.T) { + server := NewServer() + service := new(Service) + + if err := server.Register(service); err != nil { + t.Fatalf("%v", err) + } + + if len(server.services) != 2 { + t.Fatalf("Expected 2 services entries, got %d", len(server.services)) + } + + svc, ok := server.services["service"] + if !ok { + t.Fatalf("Unable to locate 'service' service after registration") + } + + if len(svc.callbacks) != 3 { + t.Errorf("Expected 3 callbacks for service 'calc', got %d", len(svc.callbacks)) + } + + if len(svc.subscriptions) != 1 { + t.Errorf("Expected 1 subscription for service 'calc', got %d", len(svc.subscriptions)) + } +} + +func TestServerRegisterName(t *testing.T) { + server := NewServer() + service := new(Service) + + if err := server.RegisterName("calc", service); err != nil { + t.Fatalf("%v", err) + } + + if len(server.services) != 2 { + t.Fatalf("Expected 2 service entries, got %d", len(server.services)) + } + + svc, ok := server.services["calc"] + if !ok { + t.Fatalf("Expected service calc to be registered") + } + + if len(svc.callbacks) != 3 { + t.Errorf("Expected 3 callbacks for service 'calc', got %d", len(svc.callbacks)) + } + + if len(svc.subscriptions) != 1 { + t.Errorf("Expected 1 subscription for service 'calc', got %d", len(svc.subscriptions)) + } +} + +// dummy codec used for testing RPC method execution +type ServerTestCodec struct { + counter int + input []byte + output string + closer chan interface{} +} + +func (c *ServerTestCodec) ReadRequestHeaders() ([]rpcRequest, bool, RPCError) { + c.counter += 1 + + if c.counter == 1 { + var req jsonRequest + json.Unmarshal(c.input, &req) + return []rpcRequest{rpcRequest{id: req.Id, isPubSub: false, service: "test", method: req.Method, params: req.Payload}}, false, nil + } + + // requests are executes in parallel, wait a bit before returning an error so that the previous request has time to + // be executed + timer := time.NewTimer(time.Duration(2) * time.Second) + <-timer.C + + return nil, false, &invalidRequestError{"connection closed"} +} + +func (c *ServerTestCodec) ParseRequestArguments(argTypes []reflect.Type, payload interface{}) ([]reflect.Value, RPCError) { + + args, _ := payload.(json.RawMessage) + + 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) + } + } + } + + if err := json.Unmarshal(args, ¶ms); err != nil { + return nil, &invalidParamsError{err.Error()} + } + + // Convert pointers back to values where necessary + for i, a := range argValues { + if a.Kind() != argTypes[i].Kind() { + argValues[i] = reflect.Indirect(argValues[i]) + } + } + + return argValues, nil +} + +func (c *ServerTestCodec) CreateResponse(id int64, reply interface{}) interface{} { + return &jsonSuccessResponse{Version: jsonRPCVersion, Id: id, Result: reply} +} + +func (c *ServerTestCodec) CreateErrorResponse(id *int64, err RPCError) interface{} { + return &jsonErrResponse{Version: jsonRPCVersion, Id: id, Error: jsonError{Code: err.Code(), Message: err.Error()}} +} + +func (c *ServerTestCodec) CreateErrorResponseWithInfo(id *int64, err RPCError, info interface{}) interface{} { + return &jsonErrResponse{Version: jsonRPCVersion, Id: id, + Error: jsonError{Code: err.Code(), Message: err.Error(), Data: info}} +} + +func (c *ServerTestCodec) CreateNotification(subid string, event interface{}) interface{} { + return &jsonNotification{Version: jsonRPCVersion, Method: notificationMethod, + Params: jsonSubscription{Subscription: subid, Result: event}} +} + +func (c *ServerTestCodec) Write(msg interface{}) error { + if len(c.output) == 0 { // only capture first response + if o, err := json.Marshal(msg); err != nil { + return err + } else { + c.output = string(o) + } + } + + return nil +} + +func (c *ServerTestCodec) Close() { + close(c.closer) +} + +func (c *ServerTestCodec) Closed() <-chan interface{} { + return c.closer +} + +func TestServerMethodExecution(t *testing.T) { + server := NewServer() + service := new(Service) + + if err := server.RegisterName("test", service); err != nil { + t.Fatalf("%v", err) + } + + req := jsonRequest{ + Method: "echo", + Version: "2.0", + Id: 12345, + } + args := []interface{}{"string arg", 1122, &Args{"qwerty"}} + req.Payload, _ = json.Marshal(&args) + + input, _ := json.Marshal(&req) + codec := &ServerTestCodec{input: input, closer: make(chan interface{})} + go server.ServeCodec(codec) + + <-codec.closer + + expected := `{"jsonrpc":"2.0","id":12345,"result":{"String":"string arg","Int":1122,"Args":{"S":"qwerty"}}}` + + if expected != codec.output { + t.Fatalf("expected %s, got %s\n", expected, codec.output) + } +} diff --git a/rpc/v2/types.go b/rpc/v2/types.go new file mode 100644 index 0000000000..7ab3d91cac --- /dev/null +++ b/rpc/v2/types.go @@ -0,0 +1,340 @@ +// Copyright 2015 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 v2 + +import ( + "fmt" + "math" + "math/big" + "reflect" + "strings" + + "sync" + + "github.com/ethereum/go-ethereum/event" +) + +// 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 + isSubscribe bool // indication if the callback is a subscription +} + +// service represents a registered object +type service struct { + name string // name for service + rcvr reflect.Value // receiver of methods for the service + typ reflect.Type // receiver type + callbacks callbacks // registered handlers + subscriptions subscriptions // available subscriptions/notifications +} + +// serverRequest is an incoming request +type serverRequest struct { + id int64 + svcname string + rcvr reflect.Value + callb *callback + args []reflect.Value + isUnsubscribe bool + err RPCError +} + +type serviceRegistry map[string]*service // collection of services +type callbacks map[string]*callback // collection of RPC callbacks +type subscriptions map[string]*callback // collection of subscription callbacks +type subscriptionRegistry map[string]Subscription // collection of subscriptions + +// Server represents a RPC server +type Server struct { + services serviceRegistry + muSubcriptions sync.Mutex // protects subscriptions + subscriptions subscriptionRegistry +} + +// rpcRequest represents a raw incoming RPC request +type rpcRequest struct { + service string + method string + id int64 + isPubSub bool + params interface{} +} + +// RPCError implements RPC error, is add support for error codec over regular go errors +type RPCError interface { + // RPC error code + Code() int + // Error message + Error() string +} + +// ServerCodec implements reading, parsing and writing RPC messages for the server side of +// a RPC session. Implementations must be go-routine safe since the codec can be called in +// multiple go-routines concurrently. +type ServerCodec interface { + // Read next request + ReadRequestHeaders() ([]rpcRequest, bool, RPCError) + // Parse request argument to the given types + ParseRequestArguments([]reflect.Type, interface{}) ([]reflect.Value, RPCError) + // Assemble success response + CreateResponse(int64, interface{}) interface{} + // Assemble error response + CreateErrorResponse(*int64, RPCError) interface{} + // Assemble error response with extra information about the error through info + CreateErrorResponseWithInfo(id *int64, err RPCError, info interface{}) interface{} + // Create notification response + CreateNotification(string, interface{}) interface{} + // Write msg to client. + Write(interface{}) error + // Close underlying data stream + Close() + // Closed when underlying connection is closed + Closed() <-chan 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 + format SubscriptionOutputFormat +} + +// NewSubscription create a new RPC subscription +func NewSubscription(sub event.Subscription) Subscription { + 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, defaultSubscriptionOutputFormatter} +} + +// Chan returns the channel where new events will be published. It's up the user to call the matcher to +// determine if the events are interesting for the client. +func (s *Subscription) Chan() <-chan *event.Event { + return s.sub.Chan() +} + +// Unsubscribe will end the subscription and closes the event channel +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 new file mode 100644 index 0000000000..2bbb9c8259 --- /dev/null +++ b/rpc/v2/utils.go @@ -0,0 +1,206 @@ +// Copyright 2015 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 v2 + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "math/big" + "reflect" + "unicode" + "unicode/utf8" +) + +// Is this an exported - upper case - name? +func isExported(name string) bool { + rune, _ := utf8.DecodeRuneInString(name) + return unicode.IsUpper(rune) +} + +// Is this type exported or a builtin? +func isExportedOrBuiltinType(t reflect.Type) bool { + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + // PkgPath will be non-empty even for an exported type, + // so we need to check the type name as well. + return isExported(t.Name()) || t.PkgPath() == "" +} + +var errorType = reflect.TypeOf((*error)(nil)).Elem() + +// Implements this type the error interface +func isErrorType(t reflect.Type) bool { + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + return t.Implements(errorType) +} + +var subscriptionType = reflect.TypeOf((*Subscription)(nil)).Elem() + +func isSubscriptionType(t reflect.Type) bool { + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + return t == subscriptionType +} + +// isPubSub tests whether the given method return the pair +// (v2.Subscription, error) +func isPubSub(methodType reflect.Type) bool { + if methodType.NumOut() != 2 { + return false + } + + 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(rcvr reflect.Value, typ reflect.Type) (callbacks, subscriptions) { + callbacks := make(callbacks) + subscriptions := make(subscriptions) + +METHODS: + for m := 0; m < typ.NumMethod(); m++ { + method := typ.Method(m) + mtype := method.Type + mname := formatName(method.Name) + if method.PkgPath != "" { // method must be exported + continue + } + + var h callback + h.isSubscribe = isPubSub(mtype) + h.rcvr = rcvr + h.method = method + h.errPos = -1 + + if h.isSubscribe { + h.argTypes = make([]reflect.Type, mtype.NumIn()-1) // skip rcvr type + for i := 1; i < mtype.NumIn(); i++ { + argType := mtype.In(i) + if isExportedOrBuiltinType(argType) { + h.argTypes[i-1] = argType + } else { + continue METHODS + } + } + + subscriptions[mname] = &h + continue METHODS + } + + numIn := mtype.NumIn() + + // determine method arguments, ignore first arg since it's the receiver type + // Arguments must be exported or builtin + h.argTypes = make([]reflect.Type, numIn-1) + for i := 1; i < numIn; i++ { + argType := mtype.In(i) + if !isExportedOrBuiltinType(argType) { + continue METHODS + } + h.argTypes[i-1] = argType + } + + // check that all returned values are exported or builtin values + for i := 0; i < mtype.NumOut(); i++ { + if !isExportedOrBuiltinType(mtype.Out(i)) { + continue METHODS + } + } + + // when a method returns an error it must be the last returned value + h.errPos = -1 + for i := 0; i < mtype.NumOut(); i++ { + if isErrorType(mtype.Out(i)) { + h.errPos = i + break + } + } + + if h.errPos >= 0 && h.errPos != mtype.NumOut()-1 { + continue METHODS + } + + switch mtype.NumOut() { + case 0, 1: + break + case 2: + if h.errPos == -1 { // method must one return value and 1 error + continue METHODS + } + break + default: + continue METHODS + } + callbacks[mname] = &h + } + + return callbacks, subscriptions +} + +func newSubscriptionId() (string, error) { + var subid [16]byte + n, _ := rand.Read(subid[:]) + if n != 16 { + return "", errors.New("Unable to generate subscription id") + } + return "0x" + hex.EncodeToString(subid[:]), nil +} diff --git a/whisper/whisper.go b/whisper/whisper.go index 7201062b81..2037538734 100644 --- a/whisper/whisper.go +++ b/whisper/whisper.go @@ -28,6 +28,7 @@ import ( "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" + rpc "github.com/ethereum/go-ethereum/rpc/v2" "gopkg.in/fatih/set.v0" ) @@ -103,6 +104,12 @@ func (self *Whisper) Protocols() []p2p.Protocol { return []p2p.Protocol{self.protocol} } +// Apis implements node.Servie, returning all the API handlers the service wants +// to expose. +func (self *Whisper) Apis() (string, []rpc.Api) { + return "whisper", []rpc.Api{} +} + // Version returns the whisper sub-protocols version number. func (self *Whisper) Version() uint { return self.protocol.Version diff --git a/whisper/whisper_rpc.go b/whisper/whisper_rpc.go new file mode 100644 index 0000000000..4f637e4367 --- /dev/null +++ b/whisper/whisper_rpc.go @@ -0,0 +1,396 @@ +package whisper + +import ( + rpc "github.com/ethereum/go-ethereum/rpc/v2" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "time" + "fmt" + "encoding/json" + "sync" +) + +// WhisperService provides the whisper RPC service. +type WhisperService struct { + w *Whisper + + messagesMu sync.RWMutex + messages map[int]*whisperFilter +} + +type whisperOfflineError struct{} + +func (e *whisperOfflineError) Error() string { + return "whisper is offline" +} + +// whisperOffLineErr is returned when the node doesn't offer the shh service. +var whisperOffLineErr = new(whisperOfflineError) + +// NewWhisperService create a new RPC whisper service. +func NewWhisperService(w *Whisper) *WhisperService { + return &WhisperService{w: w, messages: make(map[int]*whisperFilter)} +} + +// Version returns the Whisper version this node offers. +func (s *WhisperService) Version() (*rpc.HexNumber, error) { + if s.w == nil { + return rpc.NewHexNumber(0), whisperOffLineErr + } + return rpc.NewHexNumber(s.w.Version()), nil +} + +// HasIdentity checks if the the whisper node is configured with the private key +// of the specified public pair. +func (s *WhisperService) HasIdentity(identity string) (bool, error) { + if s.w == nil { + return false, whisperOffLineErr + } + return s.w.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil +} + +// NewIdentity generates a new cryptographic identity for the client, and injects +// it into the known identities for message decryption. +func (s *WhisperService) NewIdentity() (string, error) { + if s.w == nil { + return "", whisperOffLineErr + } + + identity := s.w.NewIdentity() + return common.ToHex(crypto.FromECDSAPub(&identity.PublicKey)), nil +} + +type NewFilterArgs struct { + To string + From string + Topics [][][]byte +} + +// NewWhisperFilter creates and registers a new message filter to watch for inbound whisper messages. +func (s *WhisperService) NewFilter(args *NewFilterArgs) (*rpc.HexNumber, error) { + if s.w == nil { + return nil, whisperOffLineErr + } + + var id int + filter := Filter{ + To: crypto.ToECDSAPub(common.FromHex(args.To)), + From: crypto.ToECDSAPub(common.FromHex(args.From)), + Topics: NewFilterTopics(args.Topics...), + Fn: func(message *Message) { + wmsg := NewWhisperMessage(message) + s.messagesMu.RLock() // Only read lock to the filter pool + defer s.messagesMu.RUnlock() + if s.messages[id] != nil { + s.messages[id].insert(wmsg) + } + }, + } + + id = s.w.Watch(filter) + + s.messagesMu.Lock() + s.messages[id] = newWhisperFilter(id, s.w) + s.messagesMu.Unlock() + + return rpc.NewHexNumber(id), nil +} + +// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval. +func (s *WhisperService) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage { + s.messagesMu.RLock() + defer s.messagesMu.RUnlock() + + if s.messages[filterId.Int()] != nil { + if changes := s.messages[filterId.Int()].retrieve(); changes != nil { + return changes + } + } + return returnWhisperMessages(nil) +} + +// UninstallFilter disables and removes an existing filter. +func (s *WhisperService) UninstallFilter(filterId rpc.HexNumber) bool { + s.messagesMu.Lock() + defer s.messagesMu.Unlock() + + if _, ok := s.messages[filterId.Int()]; ok { + delete(s.messages, filterId.Int()) + return true + } + return false +} + +// GetMessages retrieves all the known messages that match a specific filter. +func (s *WhisperService) GetMessages(filterId rpc.HexNumber) []WhisperMessage { + // Retrieve all the cached messages matching a specific, existing filter + s.messagesMu.RLock() + defer s.messagesMu.RUnlock() + + var messages []*Message + if s.messages[filterId.Int()] != nil { + messages = s.messages[filterId.Int()].messages() + } + + return returnWhisperMessages(messages) +} + +// returnWhisperMessages converts aNhisper message to a RPC whisper message. +func returnWhisperMessages(messages []*Message) []WhisperMessage { + msgs := make([]WhisperMessage, len(messages)) + for i, msg := range messages { + msgs[i] = NewWhisperMessage(msg) + } + return msgs +} + +type PostArgs struct { + From string `json:"from"` + To string `json:"to"` + Topics [][]byte `json:"topics"` + Payload string `json:"payload"` + Priority int64 `json:"priority"` + TTL int64 `json:"ttl"` +} + +// Post injects a message into the whisper network for distribution. +func (s *WhisperService) Post(args *PostArgs) (bool, error) { + if s.w == nil { + return false, whisperOffLineErr + } + + // construct whisper message with transmission options + message := NewMessage(common.FromHex(args.Payload)) + options := Options{ + To: crypto.ToECDSAPub(common.FromHex(args.To)), + TTL: time.Duration(args.TTL) * time.Second, + Topics: NewTopics(args.Topics...), + } + + // set sender identity + if len(args.From) > 0 { + if key := s.w.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); key != nil { + options.From = key + } else { + return false, fmt.Errorf("unknown identity to send from: %s", args.From) + } + } + + // Wrap and send the message + pow := time.Duration(args.Priority) * time.Millisecond + envelope, err := message.Wrap(pow, options) + if err != nil { + return false, err + } + + return true, s.w.Send(envelope) +} + +// WhisperMessage is the RPC representation of a whisper message. +type WhisperMessage struct { + ref *Message + + Payload string `json:"payload"` + To string `json:"to"` + From string `json:"from"` + Sent int64 `json:"sent"` + TTL int64 `json:"ttl"` + Hash string `json:"hash"` +} + +func (args *PostArgs) UnmarshalJSON(data []byte) (err error) { + var obj struct { + From string `json:"from"` + To string `json:"to"` + Topics []string `json:"topics"` + Payload string `json:"payload"` + Priority rpc.HexNumber `json:"priority"` + TTL rpc.HexNumber `json:"ttl"` + } + + if err := json.Unmarshal(data, &obj); err != nil { + return err + } + + args.From = obj.From + args.To = obj.To + args.Payload = obj.Payload + args.Priority = obj.Priority.Int64() + args.TTL = obj.TTL.Int64() + + // decode topic strings + args.Topics = make([][]byte, len(obj.Topics)) + for i, topic := range obj.Topics { + args.Topics[i] = common.FromHex(topic) + } + + return nil +} + +// UnmarshalJSON implements the json.Unmarshaler interface, invoked to convert a +// JSON message blob into a WhisperFilterArgs structure. +func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { + // Unmarshal the JSON message and sanity check + var obj struct { + To interface{} `json:"to"` + From interface{} `json:"from"` + Topics interface{} `json:"topics"` + } + if err := json.Unmarshal(b, &obj); err != nil { + return err + } + + // Retrieve the simple data contents of the filter arguments + if obj.To == nil { + args.To = "" + } else { + argstr, ok := obj.To.(string) + if !ok { + return fmt.Errorf("to is not a string") + } + args.To = argstr + } + if obj.From == nil { + args.From = "" + } else { + argstr, ok := obj.From.(string) + if !ok { + return fmt.Errorf("from is not a string") + } + args.From = argstr + } + // Construct the nested topic array + if obj.Topics != nil { + // Make sure we have an actual topic array + list, ok := obj.Topics.([]interface{}) + if !ok { + return fmt.Errorf("topics is not an array") + } + // Iterate over each topic and handle nil, string or array + topics := make([][]string, len(list)) + for idx, field := range list { + switch value := field.(type) { + case nil: + topics[idx] = []string{} + + case string: + topics[idx] = []string{value} + + case []interface{}: + topics[idx] = make([]string, len(value)) + for i, nested := range value { + switch value := nested.(type) { + case nil: + topics[idx][i] = "" + + case string: + topics[idx][i] = value + + default: + return fmt.Errorf("topic[%d][%d] is not a string", idx, i) + } + } + default: + return fmt.Errorf("topic[%d] not a string or array", idx) + } + } + + topicsDecoded := make([][][]byte, len(topics)) + for i, condition := range topics { + topicsDecoded[i] = make([][]byte, len(condition)) + for j, topic := range condition { + topicsDecoded[i][j] = common.FromHex(topic) + } + } + + args.Topics = topicsDecoded + } + return nil +} + +// whisperFilter is the message cache matching a specific filter, accumulating +// inbound messages until the are requested by the client. +type whisperFilter struct { + id int // Filter identifier for old message retrieval + ref *Whisper // Whisper reference for old message retrieval + + cache []WhisperMessage // Cache of messages not yet polled + skip map[common.Hash]struct{} // List of retrieved messages to avoid duplication + update time.Time // Time of the last message query + + lock sync.RWMutex // Lock protecting the filter internals +} + +// messages retrieves all the cached messages from the entire pool matching the +// filter, resetting the filter's change buffer. +func (w *whisperFilter) messages() []*Message { + w.lock.Lock() + defer w.lock.Unlock() + + w.cache = nil + w.update = time.Now() + + w.skip = make(map[common.Hash]struct{}) + messages := w.ref.Messages(w.id) + for _, message := range messages { + w.skip[message.Hash] = struct{}{} + } + return messages +} + +// insert injects a new batch of messages into the filter cache. +func (w *whisperFilter) insert(messages ...WhisperMessage) { + w.lock.Lock() + defer w.lock.Unlock() + + for _, message := range messages { + if _, ok := w.skip[message.ref.Hash]; !ok { + w.cache = append(w.cache, messages...) + } + } +} + +// retrieve fetches all the cached messages from the filter. +func (w *whisperFilter) retrieve() (messages []WhisperMessage) { + w.lock.Lock() + defer w.lock.Unlock() + + messages, w.cache = w.cache, nil + w.update = time.Now() + + return +} + +// activity returns the last time instance when client requests were executed on +// the filter. +func (w *whisperFilter) activity() time.Time { + w.lock.RLock() + defer w.lock.RUnlock() + + return w.update +} + +// newWhisperFilter creates a new serialized, poll based whisper topic filter. +func newWhisperFilter(id int, ref *Whisper) *whisperFilter { + return &whisperFilter{ + id: id, + ref: ref, + + update: time.Now(), + skip: make(map[common.Hash]struct{}), + } +} + +// NewWhisperMessage converts an internal message into an API version. +func NewWhisperMessage(message *Message) WhisperMessage { + return WhisperMessage{ + ref: message, + + Payload: common.ToHex(message.Payload), + From: common.ToHex(crypto.FromECDSAPub(message.Recover())), + To: common.ToHex(crypto.FromECDSAPub(message.To)), + Sent: message.Sent.Unix(), + TTL: int64(message.TTL / time.Second), + Hash: common.ToHex(message.Hash.Bytes()), + } +}