This commit is contained in:
Péter Szilágyi 2015-12-10 11:18:36 +00:00
commit 2e1006c509
45 changed files with 4331 additions and 70 deletions

View file

@ -44,6 +44,10 @@ type Account struct {
Address common.Address Address common.Address
} }
func (acc *Account) MarshalJSON() ([]byte, error) {
return []byte(`"` + acc.Address.Hex() + `"`), nil
}
type Manager struct { type Manager struct {
keyStore crypto.KeyStore keyStore crypto.KeyStore
unlocked map[common.Address]*unlocked 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) 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 // TimedUnlock unlocks the account with the given address. The account
// stays unlocked for the duration of timeout. A timeout of 0 unlocks the account // stays unlocked for the duration of timeout. A timeout of 0 unlocks the account
// until the program exits. // until the program exits.

70
accounts/account_rpc.go Normal file
View file

@ -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
}

View file

@ -313,6 +313,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso
utils.IPCDisabledFlag, utils.IPCDisabledFlag,
utils.IPCApiFlag, utils.IPCApiFlag,
utils.IPCPathFlag, utils.IPCPathFlag,
utils.IPCExperimental,
utils.ExecFlag, utils.ExecFlag,
utils.WhisperEnabledFlag, utils.WhisperEnabledFlag,
utils.DevModeFlag, utils.DevModeFlag,
@ -398,7 +399,7 @@ func attach(ctx *cli.Context) {
client, err = comms.ClientFromEndpoint(ctx.Args().First(), codec.JSON) client, err = comms.ClientFromEndpoint(ctx.Args().First(), codec.JSON)
} else { } else {
cfg := comms.IpcConfig{ 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) client, err = comms.NewIpcClient(cfg, codec.JSON)
} }
@ -506,11 +507,6 @@ func startNode(ctx *cli.Context, stack *node.Node) {
} }
} }
// Start auxiliary services if enabled. // 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 ctx.GlobalBool(utils.RPCEnabledFlag.Name) {
if err := utils.StartRPC(stack, ctx); err != nil { if err := utils.StartRPC(stack, ctx); err != nil {
utils.Fatalf("Failed to start RPC: %v", err) 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. // from a list of preloaded passphrases, or requested interactively from the user.
func getPassPhrase(prompt string, confirmation bool, i int, passwords []string) string { func getPassPhrase(prompt string, confirmation bool, i int, passwords []string) string {
// If a list of passwords was supplied, retrieve from them // If a list of passwords was supplied, retrieve from them

View file

@ -27,7 +27,6 @@ import (
"github.com/codegangsta/cli" "github.com/codegangsta/cli"
"github.com/ethereum/go-ethereum/cmd/utils" "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"
"github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/codec"
"github.com/ethereum/go-ethereum/rpc/comms" "github.com/ethereum/go-ethereum/rpc/comms"
@ -37,7 +36,7 @@ import (
var ( var (
monitorCommandAttachFlag = cli.StringFlag{ monitorCommandAttachFlag = cli.StringFlag{
Name: "attach", Name: "attach",
Value: "ipc:" + common.DefaultIpcPath(), Value: "ipc:" + utils.IPCPathFlag.Value.Value,
Usage: "API endpoint to attach to", Usage: "API endpoint to attach to",
} }
monitorCommandRowsFlag = cli.IntFlag{ monitorCommandRowsFlag = cli.IntFlag{

View file

@ -158,6 +158,7 @@ var AppHelpFlagGroups = []flagGroup{
Flags: []cli.Flag{ Flags: []cli.Flag{
utils.WhisperEnabledFlag, utils.WhisperEnabledFlag,
utils.NatspecEnabledFlag, utils.NatspecEnabledFlag,
utils.IPCExperimental,
}, },
}, },
{ {

View file

@ -23,7 +23,6 @@ import (
"log" "log"
"math" "math"
"math/big" "math/big"
"net"
"net/http" "net/http"
"os" "os"
"path/filepath" "path/filepath"
@ -52,8 +51,6 @@ import (
"github.com/ethereum/go-ethereum/rpc/api" "github.com/ethereum/go-ethereum/rpc/api"
"github.com/ethereum/go-ethereum/rpc/codec" "github.com/ethereum/go-ethereum/rpc/codec"
"github.com/ethereum/go-ethereum/rpc/comms" "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/whisper"
"github.com/ethereum/go-ethereum/xeth" "github.com/ethereum/go-ethereum/xeth"
) )
@ -297,8 +294,12 @@ var (
} }
IPCPathFlag = DirectoryFlag{ IPCPathFlag = DirectoryFlag{
Name: "ipcpath", Name: "ipcpath",
Usage: "Filename for IPC socket/pipe", Usage: "Filename for IPC socket/pipe within the datadir (explicit paths escape it)",
Value: DirectoryString{common.DefaultIpcPath()}, Value: DirectoryString{"geth.ipc"},
}
IPCExperimental = cli.BoolFlag{
Name: "ipcexp",
Usage: "Enable the new RPC implementation",
} }
ExecFlag = cli.StringFlag{ ExecFlag = cli.StringFlag{
Name: "exec", Name: "exec",
@ -405,6 +406,15 @@ func MustMakeDataDir(ctx *cli.Context) string {
return "" 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 // 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 // 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. // 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 // Configure the node's service container
stackConf := &node.Config{ stackConf := &node.Config{
DataDir: MustMakeDataDir(ctx), DataDir: MustMakeDataDir(ctx),
IpcPath: MakeIpcPath(ctx),
PrivateKey: MakeNodeKey(ctx), PrivateKey: MakeNodeKey(ctx),
Name: MakeNodeName(name, version, ctx), Name: MakeNodeName(name, version, ctx),
NoDiscovery: ctx.GlobalBool(NoDiscoverFlag.Name), NoDiscovery: ctx.GlobalBool(NoDiscoverFlag.Name),
@ -754,47 +765,6 @@ func MakeChain(ctx *cli.Context) (chain *core.BlockChain, chainDb ethdb.Database
return chain, chainDb 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(&ethereum); 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. // StartRPC starts a HTTP JSON-RPC API server.
func StartRPC(stack *node.Node, ctx *cli.Context) error { func StartRPC(stack *node.Node, ctx *cli.Context) error {
config := comms.HttpConfig{ config := comms.HttpConfig{

View file

@ -87,10 +87,3 @@ func DefaultDataDir() string {
// As we cannot guess a stable location, return empty and handle later // As we cannot guess a stable location, return empty and handle later
return "" return ""
} }
func DefaultIpcPath() string {
if runtime.GOOS == "windows" {
return `\\.\pipe\geth.ipc`
}
return filepath.Join(DefaultDataDir(), "geth.ipc")
}

View file

@ -17,6 +17,8 @@
package common package common
import ( import (
"encoding/hex"
"encoding/json"
"fmt" "fmt"
"math/big" "math/big"
"math/rand" "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) Big() *big.Int { return Bytes2Big(h[:]) }
func (h Hash) Hex() string { return "0x" + Bytes2Hex(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 // Sets the hash to the value of b. If b is larger than len(h) it will panic
func (h *Hash) SetBytes(b []byte) { func (h *Hash) SetBytes(b []byte) {
if len(b) > len(h) { 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: // PP Pretty Prints a byte slice in the following format:
// hex(value[:4])...(hex[len(value)-4:]) // hex(value[:4])...(hex[len(value)-4:])
func PP(value []byte) string { func PP(value []byte) string {

329
core/blockchain_rpc.go Normal file
View file

@ -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
}

View file

@ -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
}

View file

@ -48,6 +48,10 @@ func (n BlockNonce) Uint64() uint64 {
return binary.BigEndian.Uint64(n[:]) return binary.BigEndian.Uint64(n[:])
} }
func (n BlockNonce) MarshalJSON() ([]byte, error) {
return []byte(fmt.Sprintf(`"0x%x"`, n)), nil
}
type Header struct { type Header struct {
ParentHash common.Hash // Hash to the previous block ParentHash common.Hash // Hash to the previous block
UncleHash common.Hash // Uncles of this block UncleHash common.Hash // Uncles of this block

View file

@ -69,6 +69,10 @@ func (b Bloom) TestBytes(test []byte) bool {
return b.Test(common.BytesToBig(test)) 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 { func CreateBloom(receipts Receipts) Bloom {
bin := new(big.Int) bin := new(big.Int)
for _, receipt := range receipts { for _, receipt := range receipts {

View file

@ -17,11 +17,13 @@
package vm package vm
import ( import (
"encoding/json"
"fmt" "fmt"
"io" "io"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
rpc "github.com/ethereum/go-ethereum/rpc/v2"
) )
type Log struct { 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) 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 type Logs []*Log
// LogForStorage is a wrapper around a Log that flattens and parses the entire // LogForStorage is a wrapper around a Log that flattens and parses the entire

View file

@ -43,6 +43,7 @@ import (
"github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
rpc "github.com/ethereum/go-ethereum/rpc/v2"
) )
const ( const (
@ -282,6 +283,12 @@ func (s *Ethereum) Protocols() []p2p.Protocol {
return s.protocolManager.SubProtocols 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 // Start implements node.Service, starting all internal goroutines needed by the
// Ethereum protocol implementation. // Ethereum protocol implementation.
func (s *Ethereum) Start(*p2p.Server) error { func (s *Ethereum) Start(*p2p.Server) error {

View file

@ -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
}

76
eth/eth_rpc.go Normal file
View file

@ -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
}

548
eth/filters/filter_rpc.go Normal file
View file

@ -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
}

23
eth/web3_rpc.go Normal file
View file

@ -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)))
}

52
miner/miner_rpc.go Normal file
View file

@ -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
}

View file

@ -327,6 +327,7 @@ func (self *worker) wait() {
go func(block *types.Block, logs vm.Logs, receipts []*types.Receipt) { go func(block *types.Block, logs vm.Logs, receipts []*types.Receipt) {
self.mux.Post(core.NewMinedBlockEvent{block}) self.mux.Post(core.NewMinedBlockEvent{block})
self.mux.Post(core.ChainEvent{block, block.Hash(), logs}) self.mux.Post(core.ChainEvent{block, block.Hash(), logs})
if stat == core.CanonStatTy { if stat == core.CanonStatTy {
self.mux.Post(core.ChainHeadEvent{block}) self.mux.Post(core.ChainHeadEvent{block})
self.mux.Post(logs) self.mux.Post(logs)

22
node/api.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
package node
// PublicApi is the collection of public API methods exposed by the node.
type PublicApi struct {
node *Node
}

View file

@ -23,6 +23,8 @@ import (
"net" "net"
"os" "os"
"path/filepath" "path/filepath"
"runtime"
"strings"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger"
@ -49,6 +51,12 @@ type Config struct {
// in memory. // in memory.
DataDir string 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 // 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 // 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 // is configured, the preset one is loaded from the data dir, generating it if
@ -90,6 +98,31 @@ type Config struct {
MaxPendingPeers int 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 // 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 // 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. // data folder. If no key can be found, a new one is generated.

View file

@ -21,6 +21,7 @@ import (
"io/ioutil" "io/ioutil"
"os" "os"
"path/filepath" "path/filepath"
"runtime"
"testing" "testing"
"github.com/ethereum/go-ethereum/crypto" "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 // Tests that node keys can be correctly created, persisted, loaded and/or made
// ephemeral. // ephemeral.
func TestNodeKeyPersistency(t *testing.T) { func TestNodeKeyPersistency(t *testing.T) {

View file

@ -32,6 +32,17 @@ func (e *DuplicateServiceError) Error() string {
return fmt.Sprintf("duplicate service: %v", e.Kind) 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 // StopError is returned if a Node fails to stop either any of its registered
// services or itself. // services or itself.
type StopError struct { type StopError struct {

View file

@ -19,6 +19,8 @@ package node
import ( import (
"errors" "errors"
"fmt"
"net"
"os" "os"
"path/filepath" "path/filepath"
"reflect" "reflect"
@ -26,7 +28,11 @@ import (
"syscall" "syscall"
"github.com/ethereum/go-ethereum/event" "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/p2p"
"github.com/ethereum/go-ethereum/rpc/comms"
rpc "github.com/ethereum/go-ethereum/rpc/v2"
) )
var ( var (
@ -50,6 +56,9 @@ type Node struct {
serviceFuncs []ServiceConstructor // Service constructors (in dependency order) serviceFuncs []ServiceConstructor // Service constructors (in dependency order)
services map[reflect.Type]Service // Currently running services 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 stop chan struct{} // Channel to wait for termination notifications
lock sync.RWMutex lock sync.RWMutex
} }
@ -85,6 +94,7 @@ func New(conf *Config) (*Node, error) {
MaxPendingPeers: conf.MaxPendingPeers, MaxPendingPeers: conf.MaxPendingPeers,
}, },
serviceFuncs: []ServiceConstructor{}, serviceFuncs: []ServiceConstructor{},
ipcEndpoint: conf.IpcEndpoint(),
eventmux: new(event.TypeMux), eventmux: new(event.TypeMux),
}, nil }, nil
} }
@ -162,6 +172,14 @@ func (n *Node) Start() error {
// Mark the service started for potential cleanup // Mark the service started for potential cleanup
started = append(started, kind) 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 // Finish initializing the startup
n.services = services n.services = services
n.server = running n.server = running
@ -170,6 +188,78 @@ func (n *Node) Start() error {
return nil 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 // Stop terminates a running node along with all it's services. In the node was
// not started, an error is returned. // not started, an error is returned.
func (n *Node) Stop() error { func (n *Node) Stop() error {
@ -180,7 +270,11 @@ func (n *Node) Stop() error {
if n.server == nil { if n.server == nil {
return ErrNodeStopped 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{ failure := &StopError{
Services: make(map[reflect.Type]error), Services: make(map[reflect.Type]error),
} }

View file

@ -23,6 +23,7 @@ import (
"github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover" "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 // 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: // The following methods are needed to implement a node.Service:
// - Protocols() []p2p.Protocol - devp2p protocols the service can communicate on // - 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 // - Start() error - method invoked when the node is ready to start the service
// - Stop() error - method invoked when the node terminates the service // - Stop() error - method invoked when the node terminates the service
type SampleService struct{} type SampleService struct{}
func (s *SampleService) Protocols() []p2p.Protocol { return nil } 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) Start(*p2p.Server) error { fmt.Println("Service starting..."); return nil }
func (s *SampleService) Stop() error { fmt.Println("Service stopping..."); return nil } func (s *SampleService) Stop() error { fmt.Println("Service stopping..."); return nil }

View file

@ -18,19 +18,27 @@ package node
import ( import (
"errors" "errors"
"fmt"
"io/ioutil" "io/ioutil"
"math/rand"
"os" "os"
"reflect" "reflect"
"testing" "testing"
"time"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p" "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 ( var (
testNodeKey, _ = crypto.GenerateKey() testNodeKey, _ = crypto.GenerateKey()
testNodeConfig = &Config{ testNodeConfig = &Config{
IpcPath: fmt.Sprintf("test-%d.ipc", rand.Int63()),
PrivateKey: testNodeKey, PrivateKey: testNodeKey,
Name: "test node", 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)
}
}
}

View file

@ -23,6 +23,7 @@ import (
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
rpc "github.com/ethereum/go-ethereum/rpc/v2"
) )
// ServiceContext is a collection of service independent options inherited from // 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. // Protocol retrieves the P2P protocols the service wishes to start.
Protocols() []p2p.Protocol 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 // Start is called after all services have been constructed and the networking
// layer was also initialized to spawn any goroutines required by the service. // layer was also initialized to spawn any goroutines required by the service.
Start(server *p2p.Server) error Start(server *p2p.Server) error

View file

@ -23,12 +23,14 @@ import (
"reflect" "reflect"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
rpc "github.com/ethereum/go-ethereum/rpc/v2"
) )
// NoopService is a trivial implementation of the Service interface. // NoopService is a trivial implementation of the Service interface.
type NoopService struct{} type NoopService struct{}
func (s *NoopService) Protocols() []p2p.Protocol { return nil } 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) Start(*p2p.Server) error { return nil }
func (s *NoopService) Stop() 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 // InstrumentedService is an implementation of Service for which all interface
// methods can be instrumented both return value as well as event hook wise. // methods can be instrumented both return value as well as event hook wise.
type InstrumentedService struct { type InstrumentedService struct {
protocols []p2p.Protocol protocols []p2p.Protocol
start error apiNamespace string
stop error apiHandlers []rpc.Api
start error
stop error
protocolsHook func() protocolsHook func()
apisHook func()
startHook func(*p2p.Server) startHook func(*p2p.Server)
stopHook func() stopHook func()
} }
@ -67,6 +72,13 @@ func (s *InstrumentedService) Protocols() []p2p.Protocol {
return s.protocols 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 { func (s *InstrumentedService) Start(server *p2p.Server) error {
if s.startHook != nil { if s.startHook != nil {
s.startHook(server) s.startHook(server)
@ -115,3 +127,14 @@ func InstrumentedServiceMakerB(base ServiceConstructor) ServiceConstructor {
func InstrumentedServiceMakerC(base ServiceConstructor) ServiceConstructor { func InstrumentedServiceMakerC(base ServiceConstructor) ServiceConstructor {
return InstrumentingWrapperMaker(base, reflect.TypeOf(InstrumentedServiceC{})) 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()
}
}

29
p2p/p2p_rpc.go Normal file
View file

@ -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)
}

View file

@ -39,9 +39,11 @@ func newMergedApi(apis ...shared.EthereumApi) *MergedApi {
mergedApi.methods = make(map[string]shared.EthereumApi) mergedApi.methods = make(map[string]shared.EthereumApi)
for _, api := range apis { for _, api := range apis {
mergedApi.apis[api.Name()] = api.ApiVersion() if api != nil {
for _, method := range api.Methods() { mergedApi.apis[api.Name()] = api.ApiVersion()
mergedApi.methods[method] = api for _, method := range api.Methods() {
mergedApi.methods[method] = api
}
} }
} }
return mergedApi return mergedApi

View file

@ -191,6 +191,8 @@ func ParseApiString(apistr string, codec codec.Codec, xeth *xeth.XEth, stack *no
apis[i] = NewPersonalApi(xeth, eth, codec) apis[i] = NewPersonalApi(xeth, eth, codec)
case shared.Web3ApiName: case shared.Web3ApiName:
apis[i] = NewWeb3Api(xeth, codec) apis[i] = NewWeb3Api(xeth, codec)
case "rpc": // gives information about the RPC interface
continue
default: default:
return nil, fmt.Errorf("Unknown API '%s'", name) return nil, fmt.Errorf("Unknown API '%s'", name)
} }

View file

@ -69,13 +69,29 @@ func (self *ipcClient) SupportedModules() (map[string]string, error) {
req := shared.Request{ req := shared.Request{
Id: 1, Id: 1,
Jsonrpc: "2.0", Jsonrpc: "2.0",
Method: "modules", Method: "rpc_modules",
} }
if err := self.coder.WriteResponse(req); err != nil { if err := self.coder.WriteResponse(req); err != nil {
return nil, err 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() res, err := self.coder.ReadResponse()
if err != nil { if err != nil {
return nil, err return nil, err
@ -108,6 +124,11 @@ func StartIpc(cfg IpcConfig, codec codec.Codec, initializer InitFunc) error {
return nil 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) { func ipcLoop(cfg IpcConfig, codec codec.Codec, initializer InitFunc, l net.Listener) {
glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint) glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)
defer os.Remove(cfg.Endpoint) defer os.Remove(cfg.Endpoint)

30
rpc/v2/api.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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
}

102
rpc/v2/doc.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
/*
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

85
rpc/v2/errors.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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
}

336
rpc/v2/json.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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, &params); 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
}

73
rpc/v2/json_test.go Normal file
View file

@ -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())
}
}

480
rpc/v2/server.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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
}

244
rpc/v2/server_test.go Normal file
View file

@ -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, &params); 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)
}
}

340
rpc/v2/types.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
}

57
rpc/v2/types_test.go Normal file
View file

@ -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)
}
}

206
rpc/v2/utils.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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
}

View file

@ -28,6 +28,7 @@ import (
"github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
rpc "github.com/ethereum/go-ethereum/rpc/v2"
"gopkg.in/fatih/set.v0" "gopkg.in/fatih/set.v0"
) )
@ -103,6 +104,12 @@ func (self *Whisper) Protocols() []p2p.Protocol {
return []p2p.Protocol{self.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. // Version returns the whisper sub-protocols version number.
func (self *Whisper) Version() uint { func (self *Whisper) Version() uint {
return self.protocol.Version return self.protocol.Version

396
whisper/whisper_rpc.go Normal file
View file

@ -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()),
}
}