From 21becb0febdf11132a40ff69c6145d9babf8458d Mon Sep 17 00:00:00 2001 From: Taylor Gerring Date: Tue, 24 Feb 2015 19:05:03 +0100 Subject: Cleanup RPC packcage --- rpc/args.go | 66 +++++++------------------ rpc/message.go | 150 ++++++++++++++++++++++++++------------------------------ rpc/packages.go | 5 +- 3 files changed, 90 insertions(+), 131 deletions(-) diff --git a/rpc/args.go b/rpc/args.go index 347f60410..e839da8bf 100644 --- a/rpc/args.go +++ b/rpc/args.go @@ -19,7 +19,7 @@ func (obj *GetBlockArgs) UnmarshalJSON(b []byte) (err error) { obj.Hash = argstr return } - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } type NewTxArgs struct { @@ -57,7 +57,7 @@ func (obj *NewTxArgs) UnmarshalJSON(b []byte) (err error) { return } - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } type PushTxArgs struct { @@ -70,12 +70,12 @@ func (obj *PushTxArgs) UnmarshalJSON(b []byte) (err error) { obj.Tx = arg0 return } - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } func (a *PushTxArgs) requirementsPushTx() error { if a.Tx == "" { - return NewErrorResponse("PushTx requires a 'tx' as argument") + return NewErrorWithMessage(errArguments, "PushTx requires a 'tx' as argument") } return nil } @@ -86,14 +86,14 @@ type GetStorageArgs struct { func (obj *GetStorageArgs) UnmarshalJSON(b []byte) (err error) { if err = json.Unmarshal(b, &obj.Address); err != nil { - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } return } func (a *GetStorageArgs) requirements() error { if len(a.Address) == 0 { - return NewErrorResponse("GetStorageAt requires an 'address' value as argument") + return NewErrorWithMessage(errArguments, "GetStorageAt requires an 'address' value as argument") } return nil } @@ -109,64 +109,39 @@ func (obj *GetStateArgs) UnmarshalJSON(b []byte) (err error) { obj.Address = arg0 return } - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } func (a *GetStateArgs) requirements() error { if a.Address == "" { - return NewErrorResponse("GetStorageAt requires an 'address' value as argument") + return NewErrorWithMessage(errArguments, "GetStorageAt requires an 'address' value as argument") } if a.Key == "" { - return NewErrorResponse("GetStorageAt requires an 'key' value as argument") + return NewErrorWithMessage(errArguments, "GetStorageAt requires an 'key' value as argument") } return nil } -type GetStorageAtRes struct { - Key string `json:"key"` - Value string `json:"value"` -} - type GetTxCountArgs struct { Address string `json:"address"` } -// type GetTxCountRes struct { -// Nonce int `json:"nonce"` -// } - func (obj *GetTxCountArgs) UnmarshalJSON(b []byte) (err error) { arg0 := "" if err = json.Unmarshal(b, &arg0); err == nil { obj.Address = arg0 return } - return NewErrorResponse("Could not determine JSON parameters") + return errDecodeArgs } func (a *GetTxCountArgs) requirements() error { if a.Address == "" { - return NewErrorResponse("GetTxCountAt requires an 'address' value as argument") + return NewErrorWithMessage(errArguments, "GetTxCountAt requires an 'address' value as argument") } return nil } -// type GetPeerCountRes struct { -// PeerCount int `json:"peerCount"` -// } - -// type GetListeningRes struct { -// IsListening bool `json:"isListening"` -// } - -// type GetCoinbaseRes struct { -// Coinbase string `json:"coinbase"` -// } - -// type GetMiningRes struct { -// IsMining bool `json:"isMining"` -// } - type GetBalanceArgs struct { Address string } @@ -177,21 +152,16 @@ func (obj *GetBalanceArgs) UnmarshalJSON(b []byte) (err error) { obj.Address = arg0 return } - return NewErrorResponse("Could not determine JSON parameters") + return errDecodeArgs } func (a *GetBalanceArgs) requirements() error { if a.Address == "" { - return NewErrorResponse("GetBalanceAt requires an 'address' value as argument") + return NewErrorWithMessage(errArguments, "GetBalanceAt requires an 'address' value as argument") } return nil } -type BalanceRes struct { - Balance string `json:"balance"` - Address string `json:"address"` -} - type GetCodeAtArgs struct { Address string } @@ -202,12 +172,12 @@ func (obj *GetCodeAtArgs) UnmarshalJSON(b []byte) (err error) { obj.Address = arg0 return } - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } func (a *GetCodeAtArgs) requirements() error { if a.Address == "" { - return NewErrorResponse("GetCodeAt requires an 'address' value as argument") + return NewErrorWithMessage(errArguments, "GetCodeAt requires an 'address' value as argument") } return nil } @@ -218,7 +188,7 @@ type Sha3Args struct { func (obj *Sha3Args) UnmarshalJSON(b []byte) (err error) { if err = json.Unmarshal(b, &obj.Data); err != nil { - return NewErrorResponse(ErrorDecodeArgs) + return errDecodeArgs } return } @@ -270,10 +240,10 @@ type DbArgs struct { func (a *DbArgs) requirements() error { if len(a.Database) == 0 { - return NewErrorResponse("DbPutArgs requires an 'Database' value as argument") + return NewErrorWithMessage(errArguments, "DbPutArgs requires an 'Database' value as argument") } if len(a.Key) == 0 { - return NewErrorResponse("DbPutArgs requires an 'Key' value as argument") + return NewErrorWithMessage(errArguments, "DbPutArgs requires an 'Key' value as argument") } return nil } diff --git a/rpc/message.go b/rpc/message.go index 825ede05b..18fbec931 100644 --- a/rpc/message.go +++ b/rpc/message.go @@ -25,12 +25,12 @@ import ( "github.com/ethereum/go-ethereum/xeth" ) -const ( - ErrorArguments = "Error: Insufficient arguments" - ErrorNotImplemented = "Error: Method not implemented" - ErrorUnknown = "Error: Unknown error" - ErrorParseRequest = "Error: Could not parse request" - ErrorDecodeArgs = "Error: Could not decode arguments" +var ( + errArguments = errors.New("Error: Insufficient arguments") + errNotImplemented = errors.New("Error: Method not implemented") + errUnknown = errors.New("Error: Unknown error") + errParseRequest = errors.New("Error: Could not parse request") + errDecodeArgs = errors.New("Error: Could not decode arguments") ) type RpcRequest struct { @@ -58,76 +58,72 @@ type RpcErrorObject struct { // Data interface{} `json:"data"` } -func NewErrorResponse(msg string) error { - return errors.New(msg) -} - -func NewErrorResponseWithError(msg string, err error) error { - return fmt.Errorf("%s: %v", msg, err) +func NewErrorWithMessage(err error, msg string) error { + return fmt.Errorf("%s: %s", err.Error(), msg) } func (req *RpcRequest) ToSha3Args() (*Sha3Args, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(Sha3Args) r := bytes.NewReader(req.Params[0]) if err := json.NewDecoder(r).Decode(args); err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToGetBlockArgs() (*GetBlockArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(GetBlockArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToNewTxArgs() (*NewTxArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(NewTxArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToPushTxArgs() (*PushTxArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(PushTxArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToGetStateArgs() (*GetStateArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(GetStateArgs) @@ -135,234 +131,228 @@ func (req *RpcRequest) ToGetStateArgs() (*GetStateArgs, error) { r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToStorageAtArgs() (*GetStorageArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(GetStorageArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToGetTxCountArgs() (*GetTxCountArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(GetTxCountArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToGetBalanceArgs() (*GetBalanceArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(GetBalanceArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToGetCodeAtArgs() (*GetCodeAtArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(GetCodeAtArgs) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToBoolArgs() (bool, error) { if len(req.Params) < 1 { - return false, NewErrorResponse(ErrorArguments) + return false, errArguments } var args bool err := json.Unmarshal(req.Params[0], &args) if err != nil { - return false, NewErrorResponse(ErrorDecodeArgs) + return false, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) return args, nil } func (req *RpcRequest) ToCompileArgs() (string, error) { if len(req.Params) < 1 { - return "", NewErrorResponse(ErrorArguments) + return "", errArguments } var args string err := json.Unmarshal(req.Params[0], &args) if err != nil { - return "", NewErrorResponse(ErrorDecodeArgs) + return "", errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) return args, nil } func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } args := new(FilterOptions) r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(args) if err != nil { - return nil, NewErrorResponse(ErrorDecodeArgs) + return nil, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) return args, nil } func (req *RpcRequest) ToFilterStringArgs() (string, error) { if len(req.Params) < 1 { - return "", NewErrorResponse(ErrorArguments) + return "", errArguments } var args string err := json.Unmarshal(req.Params[0], &args) if err != nil { - return "", NewErrorResponse(ErrorDecodeArgs) + return "", errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) return args, nil } func (req *RpcRequest) ToUninstallFilterArgs() (int, error) { if len(req.Params) < 1 { - return 0, NewErrorResponse(ErrorArguments) + return 0, errArguments } var args int err := json.Unmarshal(req.Params[0], &args) if err != nil { - return 0, NewErrorResponse(ErrorDecodeArgs) + return 0, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", args, args) return args, nil } func (req *RpcRequest) ToFilterChangedArgs() (int, error) { if len(req.Params) < 1 { - return 0, NewErrorResponse(ErrorArguments) + return 0, errArguments } var id int r := bytes.NewReader(req.Params[0]) err := json.NewDecoder(r).Decode(&id) if err != nil { - return 0, NewErrorResponse(ErrorDecodeArgs) + return 0, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", id, id) return id, nil } func (req *RpcRequest) ToDbPutArgs() (*DbArgs, error) { if len(req.Params) < 3 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } var args DbArgs err := json.Unmarshal(req.Params[0], &args.Database) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } err = json.Unmarshal(req.Params[1], &args.Key) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } err = json.Unmarshal(req.Params[2], &args.Value) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } - rpclogger.DebugDetailf("%T %v", args, args) + return &args, nil } func (req *RpcRequest) ToDbGetArgs() (*DbArgs, error) { if len(req.Params) < 2 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } var args DbArgs err := json.Unmarshal(req.Params[0], &args.Database) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } err = json.Unmarshal(req.Params[1], &args.Key) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } - rpclogger.DebugDetailf("%T %v", args, args) + return &args, nil } func (req *RpcRequest) ToWhisperFilterArgs() (*xeth.Options, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } var args xeth.Options err := json.Unmarshal(req.Params[0], &args) if err != nil { - return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) } - rpclogger.DebugDetailf("%T %v", args, args) + return &args, nil } func (req *RpcRequest) ToIdArgs() (int, error) { if len(req.Params) < 1 { - return 0, NewErrorResponse(ErrorArguments) + return 0, errArguments } var id int err := json.Unmarshal(req.Params[0], &id) if err != nil { - return 0, NewErrorResponse(ErrorDecodeArgs) + return 0, errDecodeArgs } - rpclogger.DebugDetailf("%T %v", id, id) + return id, nil } func (req *RpcRequest) ToWhisperPostArgs() (*WhisperMessageArgs, error) { if len(req.Params) < 1 { - return nil, NewErrorResponse(ErrorArguments) + return nil, errArguments } var args WhisperMessageArgs @@ -370,13 +360,13 @@ func (req *RpcRequest) ToWhisperPostArgs() (*WhisperMessageArgs, error) { if err != nil { return nil, err } - rpclogger.DebugDetailf("%T %v", args, args) + return &args, nil } func (req *RpcRequest) ToWhisperHasIdentityArgs() (string, error) { if len(req.Params) < 1 { - return "", NewErrorResponse(ErrorArguments) + return "", errArguments } var args string @@ -384,13 +374,13 @@ func (req *RpcRequest) ToWhisperHasIdentityArgs() (string, error) { if err != nil { return "", err } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToRegisterArgs() (string, error) { if len(req.Params) < 1 { - return "", NewErrorResponse(ErrorArguments) + return "", errArguments } var args string @@ -398,13 +388,13 @@ func (req *RpcRequest) ToRegisterArgs() (string, error) { if err != nil { return "", err } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } func (req *RpcRequest) ToWatchTxArgs() (string, error) { if len(req.Params) < 1 { - return "", NewErrorResponse(ErrorArguments) + return "", errArguments } var args string @@ -412,6 +402,6 @@ func (req *RpcRequest) ToWatchTxArgs() (string, error) { if err != nil { return "", err } - rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } diff --git a/rpc/packages.go b/rpc/packages.go index 0f00441d2..813fef949 100644 --- a/rpc/packages.go +++ b/rpc/packages.go @@ -9,7 +9,6 @@ For each request type, define the following: package rpc import ( - "fmt" "math/big" "strings" "sync" @@ -410,7 +409,7 @@ func (p *EthereumApi) WhisperMessages(id int, reply *interface{}) error { } func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error { - // Spec at https://github.com/ethereum/wiki/wiki/Generic-ON-RPC + // Spec at https://github.com/ethereum/wiki/wiki/Generic-JSON-RPC rpclogger.DebugDetailf("%T %s", req.Params, req.Params) switch req.Method { case "eth_coinbase": @@ -595,7 +594,7 @@ func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error } return p.WhisperMessages(args, reply) default: - return NewErrorResponse(fmt.Sprintf("%v %s", ErrorNotImplemented, req.Method)) + return NewErrorWithMessage(errNotImplemented, req.Method) } rpclogger.DebugDetailf("Reply: %T %s", reply, reply) -- cgit From 43acad81b3fbee608bf19e6ba32ba27b98d9a7ff Mon Sep 17 00:00:00 2001 From: Taylor Gerring Date: Tue, 24 Feb 2015 19:11:01 +0100 Subject: Rename files --- rpc/api.go | 637 +++++++++++++++++++++++++++++++++++++++++++++++++++ rpc/api_test.go | 38 +++ rpc/message.go | 407 -------------------------------- rpc/messages.go | 407 ++++++++++++++++++++++++++++++++ rpc/packages.go | 637 --------------------------------------------------- rpc/packages_test.go | 38 --- 6 files changed, 1082 insertions(+), 1082 deletions(-) create mode 100644 rpc/api.go create mode 100644 rpc/api_test.go delete mode 100644 rpc/message.go create mode 100644 rpc/messages.go delete mode 100644 rpc/packages.go delete mode 100644 rpc/packages_test.go diff --git a/rpc/api.go b/rpc/api.go new file mode 100644 index 000000000..813fef949 --- /dev/null +++ b/rpc/api.go @@ -0,0 +1,637 @@ +/* +For each request type, define the following: + +1. RpcRequest "To" method [message.go], which does basic validation and conversion to "Args" type via json.Decoder() +2. json.Decoder() calls "UnmarshalON" defined on each "Args" struct +3. EthereumApi method, taking the "Args" type and replying with an interface to be marshalled to ON + +*/ +package rpc + +import ( + "math/big" + "strings" + "sync" + "time" + + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/event/filter" + "github.com/ethereum/go-ethereum/state" + "github.com/ethereum/go-ethereum/xeth" +) + +var ( + defaultGasPrice = big.NewInt(10000000000000) + defaultGas = big.NewInt(10000) +) + +type EthereumApi struct { + xeth *xeth.XEth + quit chan struct{} + filterManager *filter.FilterManager + + logMut sync.RWMutex + logs map[int]*logFilter + + messagesMut sync.RWMutex + messages map[int]*whisperFilter + // Register keeps a list of accounts and transaction data + regmut sync.Mutex + register map[string][]*NewTxArgs + + db ethutil.Database +} + +func NewEthereumApi(eth *xeth.XEth) *EthereumApi { + db, _ := ethdb.NewLDBDatabase("dapps") + api := &EthereumApi{ + xeth: eth, + quit: make(chan struct{}), + filterManager: filter.NewFilterManager(eth.Backend().EventMux()), + logs: make(map[int]*logFilter), + messages: make(map[int]*whisperFilter), + db: db, + } + go api.filterManager.Start() + go api.start() + + return api +} + +func (self *EthereumApi) Register(args string, reply *interface{}) error { + self.regmut.Lock() + defer self.regmut.Unlock() + + if _, ok := self.register[args]; ok { + self.register[args] = nil // register with empty + } + return nil +} + +func (self *EthereumApi) Unregister(args string, reply *interface{}) error { + self.regmut.Lock() + defer self.regmut.Unlock() + + delete(self.register, args) + + return nil +} + +func (self *EthereumApi) WatchTx(args string, reply *interface{}) error { + self.regmut.Lock() + defer self.regmut.Unlock() + + txs := self.register[args] + self.register[args] = nil + + *reply = txs + return nil +} + +func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) error { + var id int + filter := core.NewFilter(self.xeth.Backend()) + filter.SetOptions(toFilterOptions(args)) + filter.LogsCallback = func(logs state.Logs) { + self.logMut.Lock() + defer self.logMut.Unlock() + + self.logs[id].add(logs...) + } + id = self.filterManager.InstallFilter(filter) + self.logs[id] = &logFilter{timeout: time.Now()} + + *reply = id + + return nil +} + +func (self *EthereumApi) UninstallFilter(id int, reply *interface{}) error { + delete(self.logs, id) + self.filterManager.UninstallFilter(id) + *reply = true + return nil +} + +func (self *EthereumApi) NewFilterString(args string, reply *interface{}) error { + var id int + filter := core.NewFilter(self.xeth.Backend()) + + callback := func(block *types.Block) { + self.logMut.Lock() + defer self.logMut.Unlock() + + self.logs[id].add(&state.StateLog{}) + } + if args == "pending" { + filter.PendingCallback = callback + } else if args == "chain" { + filter.BlockCallback = callback + } + + id = self.filterManager.InstallFilter(filter) + self.logs[id] = &logFilter{timeout: time.Now()} + *reply = id + + return nil +} + +func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error { + self.logMut.Lock() + defer self.logMut.Unlock() + + if self.logs[id] != nil { + *reply = toLogs(self.logs[id].get()) + } + + return nil +} + +func (self *EthereumApi) Logs(id int, reply *interface{}) error { + self.logMut.Lock() + defer self.logMut.Unlock() + + filter := self.filterManager.GetFilter(id) + if filter != nil { + *reply = toLogs(filter.Find()) + } + + return nil +} + +func (self *EthereumApi) AllLogs(args *FilterOptions, reply *interface{}) error { + filter := core.NewFilter(self.xeth.Backend()) + filter.SetOptions(toFilterOptions(args)) + + *reply = toLogs(filter.Find()) + + return nil +} + +func (p *EthereumApi) GetBlock(args *GetBlockArgs, reply *interface{}) error { + // This seems a bit precarious Maybe worth splitting to discrete functions + if len(args.Hash) > 0 { + *reply = p.xeth.BlockByHash(args.Hash) + } else { + *reply = p.xeth.BlockByNumber(args.BlockNumber) + } + return nil +} + +func (p *EthereumApi) Transact(args *NewTxArgs, reply *interface{}) error { + if len(args.Gas) == 0 { + args.Gas = defaultGas.String() + } + + if len(args.GasPrice) == 0 { + args.GasPrice = defaultGasPrice.String() + } + + // TODO if no_private_key then + if _, exists := p.register[args.From]; exists { + p.register[args.From] = append(p.register[args.From], args) + } else { + result, _ := p.xeth.Transact( /* TODO specify account */ args.To, args.Value, args.Gas, args.GasPrice, args.Data) + *reply = result + } + + return nil +} + +func (p *EthereumApi) Call(args *NewTxArgs, reply *interface{}) error { + result, err := p.xeth.Call( /* TODO specify account */ args.To, args.Value, args.Gas, args.GasPrice, args.Data) + if err != nil { + return err + } + + *reply = result + return nil +} + +func (p *EthereumApi) PushTx(args *PushTxArgs, reply *interface{}) error { + err := args.requirementsPushTx() + if err != nil { + return err + } + result, _ := p.xeth.PushTx(args.Tx) + *reply = result + return nil +} + +func (p *EthereumApi) GetStateAt(args *GetStateArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + + state := p.xeth.State().SafeGet(args.Address) + + value := state.StorageString(args.Key) + var hx string + if strings.Index(args.Key, "0x") == 0 { + hx = string([]byte(args.Key)[2:]) + } else { + // Convert the incoming string (which is a bigint) into hex + i, _ := new(big.Int).SetString(args.Key, 10) + hx = ethutil.Bytes2Hex(i.Bytes()) + } + rpclogger.Debugf("GetStateAt(%s, %s)\n", args.Address, hx) + *reply = map[string]string{args.Key: value.Str()} + return nil +} + +func (p *EthereumApi) GetStorageAt(args *GetStorageArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + + *reply = p.xeth.State().SafeGet(args.Address).Storage() + return nil +} + +func (p *EthereumApi) GetPeerCount(reply *interface{}) error { + *reply = p.xeth.PeerCount() + return nil +} + +func (p *EthereumApi) GetIsListening(reply *interface{}) error { + *reply = p.xeth.IsListening() + return nil +} + +func (p *EthereumApi) GetCoinbase(reply *interface{}) error { + *reply = p.xeth.Coinbase() + return nil +} + +func (p *EthereumApi) Accounts(reply *interface{}) error { + *reply = p.xeth.Accounts() + return nil +} + +func (p *EthereumApi) GetIsMining(reply *interface{}) error { + *reply = p.xeth.IsMining() + return nil +} + +func (p *EthereumApi) SetMining(shouldmine bool, reply *interface{}) error { + *reply = p.xeth.SetMining(shouldmine) + return nil +} + +func (p *EthereumApi) BlockNumber(reply *interface{}) error { + *reply = p.xeth.Backend().ChainManager().CurrentBlock().Number() + return nil +} + +func (p *EthereumApi) GetTxCountAt(args *GetTxCountArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + *reply = p.xeth.TxCountAt(args.Address) + return nil +} + +func (p *EthereumApi) GetBalanceAt(args *GetBalanceArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + state := p.xeth.State().SafeGet(args.Address) + *reply = toHex(state.Balance().Bytes()) + return nil +} + +func (p *EthereumApi) GetCodeAt(args *GetCodeAtArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + *reply = p.xeth.CodeAt(args.Address) + return nil +} + +func (p *EthereumApi) GetCompilers(reply *interface{}) error { + c := []string{"serpent"} + *reply = c + return nil +} + +func (p *EthereumApi) CompileSerpent(script string, reply *interface{}) error { + res, err := ethutil.Compile(script, false) + if err != nil { + return err + } + *reply = res + return nil +} + +func (p *EthereumApi) Sha3(args *Sha3Args, reply *interface{}) error { + *reply = toHex(crypto.Sha3(fromHex(args.Data))) + return nil +} + +func (p *EthereumApi) DbPut(args *DbArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + + p.db.Put([]byte(args.Database+args.Key), []byte(args.Value)) + *reply = true + return nil +} + +func (p *EthereumApi) DbGet(args *DbArgs, reply *interface{}) error { + err := args.requirements() + if err != nil { + return err + } + + res, _ := p.db.Get([]byte(args.Database + args.Key)) + *reply = string(res) + return nil +} + +func (p *EthereumApi) NewWhisperIdentity(reply *interface{}) error { + *reply = p.xeth.Whisper().NewIdentity() + return nil +} + +func (p *EthereumApi) NewWhisperFilter(args *xeth.Options, reply *interface{}) error { + var id int + args.Fn = func(msg xeth.WhisperMessage) { + p.messagesMut.Lock() + defer p.messagesMut.Unlock() + p.messages[id].add(msg) // = append(p.messages[id], msg) + } + id = p.xeth.Whisper().Watch(args) + p.messages[id] = &whisperFilter{timeout: time.Now()} + *reply = id + return nil +} + +func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error { + self.messagesMut.Lock() + defer self.messagesMut.Unlock() + + if self.messages[id] != nil { + *reply = self.messages[id].get() + } + + return nil +} + +func (p *EthereumApi) WhisperPost(args *WhisperMessageArgs, reply *interface{}) error { + err := p.xeth.Whisper().Post(args.Payload, args.To, args.From, args.Topic, args.Priority, args.Ttl) + if err != nil { + return err + } + + *reply = true + return nil +} + +func (p *EthereumApi) HasWhisperIdentity(args string, reply *interface{}) error { + *reply = p.xeth.Whisper().HasIdentity(args) + return nil +} + +func (p *EthereumApi) WhisperMessages(id int, reply *interface{}) error { + *reply = p.xeth.Whisper().Messages(id) + return nil +} + +func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error { + // Spec at https://github.com/ethereum/wiki/wiki/Generic-JSON-RPC + rpclogger.DebugDetailf("%T %s", req.Params, req.Params) + switch req.Method { + case "eth_coinbase": + return p.GetCoinbase(reply) + case "eth_listening": + return p.GetIsListening(reply) + case "eth_mining": + return p.GetIsMining(reply) + case "eth_setMining": + args, err := req.ToBoolArgs() + if err != nil { + return err + } + return p.SetMining(args, reply) + case "eth_peerCount": + return p.GetPeerCount(reply) + case "eth_number": + return p.BlockNumber(reply) + case "eth_accounts": + return p.Accounts(reply) + case "eth_countAt": + args, err := req.ToGetTxCountArgs() + if err != nil { + return err + } + return p.GetTxCountAt(args, reply) + case "eth_codeAt": + args, err := req.ToGetCodeAtArgs() + if err != nil { + return err + } + return p.GetCodeAt(args, reply) + case "eth_balanceAt": + args, err := req.ToGetBalanceArgs() + if err != nil { + return err + } + return p.GetBalanceAt(args, reply) + case "eth_stateAt": + args, err := req.ToGetStateArgs() + if err != nil { + return err + } + return p.GetStateAt(args, reply) + case "eth_storageAt": + args, err := req.ToStorageAtArgs() + if err != nil { + return err + } + return p.GetStorageAt(args, reply) + case "eth_blockByNumber", "eth_blockByHash": + args, err := req.ToGetBlockArgs() + if err != nil { + return err + } + return p.GetBlock(args, reply) + case "eth_transact": + args, err := req.ToNewTxArgs() + if err != nil { + return err + } + return p.Transact(args, reply) + case "eth_call": + args, err := req.ToNewTxArgs() + if err != nil { + return err + } + return p.Call(args, reply) + case "eth_newFilter": + args, err := req.ToFilterArgs() + if err != nil { + return err + } + return p.NewFilter(args, reply) + case "eth_newFilterString": + args, err := req.ToFilterStringArgs() + if err != nil { + return err + } + return p.NewFilterString(args, reply) + case "eth_uninstallFilter": + args, err := req.ToUninstallFilterArgs() + if err != nil { + return err + } + return p.UninstallFilter(args, reply) + case "eth_changed": + args, err := req.ToIdArgs() + if err != nil { + return err + } + return p.FilterChanged(args, reply) + case "eth_filterLogs": + args, err := req.ToIdArgs() + if err != nil { + return err + } + return p.Logs(args, reply) + case "eth_logs": + args, err := req.ToFilterArgs() + if err != nil { + return err + } + return p.AllLogs(args, reply) + case "eth_gasPrice": + *reply = toHex(defaultGasPrice.Bytes()) + return nil + case "eth_register": + args, err := req.ToRegisterArgs() + if err != nil { + return err + } + return p.Register(args, reply) + case "eth_unregister": + args, err := req.ToRegisterArgs() + if err != nil { + return err + } + return p.Unregister(args, reply) + case "eth_watchTx": + args, err := req.ToWatchTxArgs() + if err != nil { + return err + } + return p.WatchTx(args, reply) + case "eth_compilers": + return p.GetCompilers(reply) + case "eth_serpent": + args, err := req.ToCompileArgs() + if err != nil { + return err + } + return p.CompileSerpent(args, reply) + case "web3_sha3": + args, err := req.ToSha3Args() + if err != nil { + return err + } + return p.Sha3(args, reply) + case "db_put": + args, err := req.ToDbPutArgs() + if err != nil { + return err + } + return p.DbPut(args, reply) + case "db_get": + args, err := req.ToDbGetArgs() + if err != nil { + return err + } + return p.DbGet(args, reply) + case "shh_newIdentity": + return p.NewWhisperIdentity(reply) + case "shh_newFilter": + args, err := req.ToWhisperFilterArgs() + if err != nil { + return err + } + return p.NewWhisperFilter(args, reply) + case "shh_changed": + args, err := req.ToIdArgs() + if err != nil { + return err + } + return p.MessagesChanged(args, reply) + case "shh_post": + args, err := req.ToWhisperPostArgs() + if err != nil { + return err + } + return p.WhisperPost(args, reply) + case "shh_haveIdentity": + args, err := req.ToWhisperHasIdentityArgs() + if err != nil { + return err + } + return p.HasWhisperIdentity(args, reply) + case "shh_getMessages": + args, err := req.ToIdArgs() + if err != nil { + return err + } + return p.WhisperMessages(args, reply) + default: + return NewErrorWithMessage(errNotImplemented, req.Method) + } + + rpclogger.DebugDetailf("Reply: %T %s", reply, reply) + return nil +} + +var filterTickerTime = 15 * time.Second + +func (self *EthereumApi) start() { + timer := time.NewTicker(filterTickerTime) +done: + for { + select { + case <-timer.C: + self.logMut.Lock() + self.messagesMut.Lock() + for id, filter := range self.logs { + if time.Since(filter.timeout) > 20*time.Second { + self.filterManager.UninstallFilter(id) + delete(self.logs, id) + } + } + + for id, filter := range self.messages { + if time.Since(filter.timeout) > 20*time.Second { + self.xeth.Whisper().Unwatch(id) + delete(self.messages, id) + } + } + self.logMut.Unlock() + self.messagesMut.Unlock() + case <-self.quit: + break done + } + } +} + +func (self *EthereumApi) stop() { + close(self.quit) +} diff --git a/rpc/api_test.go b/rpc/api_test.go new file mode 100644 index 000000000..a9fc16cd3 --- /dev/null +++ b/rpc/api_test.go @@ -0,0 +1,38 @@ +package rpc + +import ( + "sync" + "testing" + "time" +) + +func TestFilterClose(t *testing.T) { + t.Skip() + api := &EthereumApi{ + logs: make(map[int]*logFilter), + messages: make(map[int]*whisperFilter), + quit: make(chan struct{}), + } + + filterTickerTime = 1 + api.logs[0] = &logFilter{} + api.messages[0] = &whisperFilter{} + var wg sync.WaitGroup + wg.Add(1) + go api.start() + go func() { + select { + case <-time.After(500 * time.Millisecond): + api.stop() + wg.Done() + } + }() + wg.Wait() + if len(api.logs) != 0 { + t.Error("expected logs to be empty") + } + + if len(api.messages) != 0 { + t.Error("expected messages to be empty") + } +} diff --git a/rpc/message.go b/rpc/message.go deleted file mode 100644 index 18fbec931..000000000 --- a/rpc/message.go +++ /dev/null @@ -1,407 +0,0 @@ -/* - This file is part of go-ethereum - - go-ethereum is free software: you can redistribute it and/or modify - it under the terms of the GNU General Public License as published by - the Free Software Foundation, either version 3 of the License, or - (at your option) any later version. - - go-ethereum 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 General Public License for more details. - - You should have received a copy of the GNU General Public License - along with go-ethereum. If not, see . -*/ -package rpc - -import ( - "bytes" - "encoding/json" - "errors" - "fmt" - - "github.com/ethereum/go-ethereum/xeth" -) - -var ( - errArguments = errors.New("Error: Insufficient arguments") - errNotImplemented = errors.New("Error: Method not implemented") - errUnknown = errors.New("Error: Unknown error") - errParseRequest = errors.New("Error: Could not parse request") - errDecodeArgs = errors.New("Error: Could not decode arguments") -) - -type RpcRequest struct { - ID interface{} `json:"id"` - JsonRpc string `json:"jsonrpc"` - Method string `json:"method"` - Params []json.RawMessage `json:"params"` -} - -type RpcSuccessResponse struct { - ID interface{} `json:"id"` - JsonRpc string `json:"jsonrpc"` - Result interface{} `json:"result"` -} - -type RpcErrorResponse struct { - ID interface{} `json:"id"` - JsonRpc string `json:"jsonrpc"` - Error *RpcErrorObject `json:"error"` -} - -type RpcErrorObject struct { - Code int `json:"code"` - Message string `json:"message"` - // Data interface{} `json:"data"` -} - -func NewErrorWithMessage(err error, msg string) error { - return fmt.Errorf("%s: %s", err.Error(), msg) -} - -func (req *RpcRequest) ToSha3Args() (*Sha3Args, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(Sha3Args) - r := bytes.NewReader(req.Params[0]) - if err := json.NewDecoder(r).Decode(args); err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToGetBlockArgs() (*GetBlockArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(GetBlockArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToNewTxArgs() (*NewTxArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(NewTxArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - - return args, nil -} - -func (req *RpcRequest) ToPushTxArgs() (*PushTxArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(PushTxArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToGetStateArgs() (*GetStateArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(GetStateArgs) - // TODO need to pass both arguments - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToStorageAtArgs() (*GetStorageArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(GetStorageArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToGetTxCountArgs() (*GetTxCountArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(GetTxCountArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToGetBalanceArgs() (*GetBalanceArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(GetBalanceArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToGetCodeAtArgs() (*GetCodeAtArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(GetCodeAtArgs) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToBoolArgs() (bool, error) { - if len(req.Params) < 1 { - return false, errArguments - } - - var args bool - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return false, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToCompileArgs() (string, error) { - if len(req.Params) < 1 { - return "", errArguments - } - - var args string - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return "", errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - args := new(FilterOptions) - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(args) - if err != nil { - return nil, errDecodeArgs - } - return args, nil -} - -func (req *RpcRequest) ToFilterStringArgs() (string, error) { - if len(req.Params) < 1 { - return "", errArguments - } - - var args string - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return "", errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToUninstallFilterArgs() (int, error) { - if len(req.Params) < 1 { - return 0, errArguments - } - - var args int - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return 0, errDecodeArgs - } - - return args, nil -} - -func (req *RpcRequest) ToFilterChangedArgs() (int, error) { - if len(req.Params) < 1 { - return 0, errArguments - } - - var id int - r := bytes.NewReader(req.Params[0]) - err := json.NewDecoder(r).Decode(&id) - if err != nil { - return 0, errDecodeArgs - } - return id, nil -} - -func (req *RpcRequest) ToDbPutArgs() (*DbArgs, error) { - if len(req.Params) < 3 { - return nil, errArguments - } - - var args DbArgs - err := json.Unmarshal(req.Params[0], &args.Database) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - err = json.Unmarshal(req.Params[1], &args.Key) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - err = json.Unmarshal(req.Params[2], &args.Value) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - - return &args, nil -} - -func (req *RpcRequest) ToDbGetArgs() (*DbArgs, error) { - if len(req.Params) < 2 { - return nil, errArguments - } - - var args DbArgs - err := json.Unmarshal(req.Params[0], &args.Database) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - - err = json.Unmarshal(req.Params[1], &args.Key) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - - return &args, nil -} - -func (req *RpcRequest) ToWhisperFilterArgs() (*xeth.Options, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - var args xeth.Options - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) - } - - return &args, nil -} - -func (req *RpcRequest) ToIdArgs() (int, error) { - if len(req.Params) < 1 { - return 0, errArguments - } - - var id int - err := json.Unmarshal(req.Params[0], &id) - if err != nil { - return 0, errDecodeArgs - } - - return id, nil -} - -func (req *RpcRequest) ToWhisperPostArgs() (*WhisperMessageArgs, error) { - if len(req.Params) < 1 { - return nil, errArguments - } - - var args WhisperMessageArgs - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return nil, err - } - - return &args, nil -} - -func (req *RpcRequest) ToWhisperHasIdentityArgs() (string, error) { - if len(req.Params) < 1 { - return "", errArguments - } - - var args string - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return "", err - } - - return args, nil -} - -func (req *RpcRequest) ToRegisterArgs() (string, error) { - if len(req.Params) < 1 { - return "", errArguments - } - - var args string - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return "", err - } - - return args, nil -} - -func (req *RpcRequest) ToWatchTxArgs() (string, error) { - if len(req.Params) < 1 { - return "", errArguments - } - - var args string - err := json.Unmarshal(req.Params[0], &args) - if err != nil { - return "", err - } - - return args, nil -} diff --git a/rpc/messages.go b/rpc/messages.go new file mode 100644 index 000000000..18fbec931 --- /dev/null +++ b/rpc/messages.go @@ -0,0 +1,407 @@ +/* + This file is part of go-ethereum + + go-ethereum is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + go-ethereum 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 General Public License for more details. + + You should have received a copy of the GNU General Public License + along with go-ethereum. If not, see . +*/ +package rpc + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + + "github.com/ethereum/go-ethereum/xeth" +) + +var ( + errArguments = errors.New("Error: Insufficient arguments") + errNotImplemented = errors.New("Error: Method not implemented") + errUnknown = errors.New("Error: Unknown error") + errParseRequest = errors.New("Error: Could not parse request") + errDecodeArgs = errors.New("Error: Could not decode arguments") +) + +type RpcRequest struct { + ID interface{} `json:"id"` + JsonRpc string `json:"jsonrpc"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` +} + +type RpcSuccessResponse struct { + ID interface{} `json:"id"` + JsonRpc string `json:"jsonrpc"` + Result interface{} `json:"result"` +} + +type RpcErrorResponse struct { + ID interface{} `json:"id"` + JsonRpc string `json:"jsonrpc"` + Error *RpcErrorObject `json:"error"` +} + +type RpcErrorObject struct { + Code int `json:"code"` + Message string `json:"message"` + // Data interface{} `json:"data"` +} + +func NewErrorWithMessage(err error, msg string) error { + return fmt.Errorf("%s: %s", err.Error(), msg) +} + +func (req *RpcRequest) ToSha3Args() (*Sha3Args, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(Sha3Args) + r := bytes.NewReader(req.Params[0]) + if err := json.NewDecoder(r).Decode(args); err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToGetBlockArgs() (*GetBlockArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(GetBlockArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToNewTxArgs() (*NewTxArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(NewTxArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + + return args, nil +} + +func (req *RpcRequest) ToPushTxArgs() (*PushTxArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(PushTxArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToGetStateArgs() (*GetStateArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(GetStateArgs) + // TODO need to pass both arguments + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToStorageAtArgs() (*GetStorageArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(GetStorageArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToGetTxCountArgs() (*GetTxCountArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(GetTxCountArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToGetBalanceArgs() (*GetBalanceArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(GetBalanceArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToGetCodeAtArgs() (*GetCodeAtArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(GetCodeAtArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToBoolArgs() (bool, error) { + if len(req.Params) < 1 { + return false, errArguments + } + + var args bool + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return false, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToCompileArgs() (string, error) { + if len(req.Params) < 1 { + return "", errArguments + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return "", errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + args := new(FilterOptions) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, errDecodeArgs + } + return args, nil +} + +func (req *RpcRequest) ToFilterStringArgs() (string, error) { + if len(req.Params) < 1 { + return "", errArguments + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return "", errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToUninstallFilterArgs() (int, error) { + if len(req.Params) < 1 { + return 0, errArguments + } + + var args int + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return 0, errDecodeArgs + } + + return args, nil +} + +func (req *RpcRequest) ToFilterChangedArgs() (int, error) { + if len(req.Params) < 1 { + return 0, errArguments + } + + var id int + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(&id) + if err != nil { + return 0, errDecodeArgs + } + return id, nil +} + +func (req *RpcRequest) ToDbPutArgs() (*DbArgs, error) { + if len(req.Params) < 3 { + return nil, errArguments + } + + var args DbArgs + err := json.Unmarshal(req.Params[0], &args.Database) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + err = json.Unmarshal(req.Params[1], &args.Key) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + err = json.Unmarshal(req.Params[2], &args.Value) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + + return &args, nil +} + +func (req *RpcRequest) ToDbGetArgs() (*DbArgs, error) { + if len(req.Params) < 2 { + return nil, errArguments + } + + var args DbArgs + err := json.Unmarshal(req.Params[0], &args.Database) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + + err = json.Unmarshal(req.Params[1], &args.Key) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + + return &args, nil +} + +func (req *RpcRequest) ToWhisperFilterArgs() (*xeth.Options, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + var args xeth.Options + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return nil, NewErrorWithMessage(errDecodeArgs, err.Error()) + } + + return &args, nil +} + +func (req *RpcRequest) ToIdArgs() (int, error) { + if len(req.Params) < 1 { + return 0, errArguments + } + + var id int + err := json.Unmarshal(req.Params[0], &id) + if err != nil { + return 0, errDecodeArgs + } + + return id, nil +} + +func (req *RpcRequest) ToWhisperPostArgs() (*WhisperMessageArgs, error) { + if len(req.Params) < 1 { + return nil, errArguments + } + + var args WhisperMessageArgs + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return nil, err + } + + return &args, nil +} + +func (req *RpcRequest) ToWhisperHasIdentityArgs() (string, error) { + if len(req.Params) < 1 { + return "", errArguments + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return "", err + } + + return args, nil +} + +func (req *RpcRequest) ToRegisterArgs() (string, error) { + if len(req.Params) < 1 { + return "", errArguments + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return "", err + } + + return args, nil +} + +func (req *RpcRequest) ToWatchTxArgs() (string, error) { + if len(req.Params) < 1 { + return "", errArguments + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return "", err + } + + return args, nil +} diff --git a/rpc/packages.go b/rpc/packages.go deleted file mode 100644 index 813fef949..000000000 --- a/rpc/packages.go +++ /dev/null @@ -1,637 +0,0 @@ -/* -For each request type, define the following: - -1. RpcRequest "To" method [message.go], which does basic validation and conversion to "Args" type via json.Decoder() -2. json.Decoder() calls "UnmarshalON" defined on each "Args" struct -3. EthereumApi method, taking the "Args" type and replying with an interface to be marshalled to ON - -*/ -package rpc - -import ( - "math/big" - "strings" - "sync" - "time" - - "github.com/ethereum/go-ethereum/core" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/ethutil" - "github.com/ethereum/go-ethereum/event/filter" - "github.com/ethereum/go-ethereum/state" - "github.com/ethereum/go-ethereum/xeth" -) - -var ( - defaultGasPrice = big.NewInt(10000000000000) - defaultGas = big.NewInt(10000) -) - -type EthereumApi struct { - xeth *xeth.XEth - quit chan struct{} - filterManager *filter.FilterManager - - logMut sync.RWMutex - logs map[int]*logFilter - - messagesMut sync.RWMutex - messages map[int]*whisperFilter - // Register keeps a list of accounts and transaction data - regmut sync.Mutex - register map[string][]*NewTxArgs - - db ethutil.Database -} - -func NewEthereumApi(eth *xeth.XEth) *EthereumApi { - db, _ := ethdb.NewLDBDatabase("dapps") - api := &EthereumApi{ - xeth: eth, - quit: make(chan struct{}), - filterManager: filter.NewFilterManager(eth.Backend().EventMux()), - logs: make(map[int]*logFilter), - messages: make(map[int]*whisperFilter), - db: db, - } - go api.filterManager.Start() - go api.start() - - return api -} - -func (self *EthereumApi) Register(args string, reply *interface{}) error { - self.regmut.Lock() - defer self.regmut.Unlock() - - if _, ok := self.register[args]; ok { - self.register[args] = nil // register with empty - } - return nil -} - -func (self *EthereumApi) Unregister(args string, reply *interface{}) error { - self.regmut.Lock() - defer self.regmut.Unlock() - - delete(self.register, args) - - return nil -} - -func (self *EthereumApi) WatchTx(args string, reply *interface{}) error { - self.regmut.Lock() - defer self.regmut.Unlock() - - txs := self.register[args] - self.register[args] = nil - - *reply = txs - return nil -} - -func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) error { - var id int - filter := core.NewFilter(self.xeth.Backend()) - filter.SetOptions(toFilterOptions(args)) - filter.LogsCallback = func(logs state.Logs) { - self.logMut.Lock() - defer self.logMut.Unlock() - - self.logs[id].add(logs...) - } - id = self.filterManager.InstallFilter(filter) - self.logs[id] = &logFilter{timeout: time.Now()} - - *reply = id - - return nil -} - -func (self *EthereumApi) UninstallFilter(id int, reply *interface{}) error { - delete(self.logs, id) - self.filterManager.UninstallFilter(id) - *reply = true - return nil -} - -func (self *EthereumApi) NewFilterString(args string, reply *interface{}) error { - var id int - filter := core.NewFilter(self.xeth.Backend()) - - callback := func(block *types.Block) { - self.logMut.Lock() - defer self.logMut.Unlock() - - self.logs[id].add(&state.StateLog{}) - } - if args == "pending" { - filter.PendingCallback = callback - } else if args == "chain" { - filter.BlockCallback = callback - } - - id = self.filterManager.InstallFilter(filter) - self.logs[id] = &logFilter{timeout: time.Now()} - *reply = id - - return nil -} - -func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error { - self.logMut.Lock() - defer self.logMut.Unlock() - - if self.logs[id] != nil { - *reply = toLogs(self.logs[id].get()) - } - - return nil -} - -func (self *EthereumApi) Logs(id int, reply *interface{}) error { - self.logMut.Lock() - defer self.logMut.Unlock() - - filter := self.filterManager.GetFilter(id) - if filter != nil { - *reply = toLogs(filter.Find()) - } - - return nil -} - -func (self *EthereumApi) AllLogs(args *FilterOptions, reply *interface{}) error { - filter := core.NewFilter(self.xeth.Backend()) - filter.SetOptions(toFilterOptions(args)) - - *reply = toLogs(filter.Find()) - - return nil -} - -func (p *EthereumApi) GetBlock(args *GetBlockArgs, reply *interface{}) error { - // This seems a bit precarious Maybe worth splitting to discrete functions - if len(args.Hash) > 0 { - *reply = p.xeth.BlockByHash(args.Hash) - } else { - *reply = p.xeth.BlockByNumber(args.BlockNumber) - } - return nil -} - -func (p *EthereumApi) Transact(args *NewTxArgs, reply *interface{}) error { - if len(args.Gas) == 0 { - args.Gas = defaultGas.String() - } - - if len(args.GasPrice) == 0 { - args.GasPrice = defaultGasPrice.String() - } - - // TODO if no_private_key then - if _, exists := p.register[args.From]; exists { - p.register[args.From] = append(p.register[args.From], args) - } else { - result, _ := p.xeth.Transact( /* TODO specify account */ args.To, args.Value, args.Gas, args.GasPrice, args.Data) - *reply = result - } - - return nil -} - -func (p *EthereumApi) Call(args *NewTxArgs, reply *interface{}) error { - result, err := p.xeth.Call( /* TODO specify account */ args.To, args.Value, args.Gas, args.GasPrice, args.Data) - if err != nil { - return err - } - - *reply = result - return nil -} - -func (p *EthereumApi) PushTx(args *PushTxArgs, reply *interface{}) error { - err := args.requirementsPushTx() - if err != nil { - return err - } - result, _ := p.xeth.PushTx(args.Tx) - *reply = result - return nil -} - -func (p *EthereumApi) GetStateAt(args *GetStateArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - - state := p.xeth.State().SafeGet(args.Address) - - value := state.StorageString(args.Key) - var hx string - if strings.Index(args.Key, "0x") == 0 { - hx = string([]byte(args.Key)[2:]) - } else { - // Convert the incoming string (which is a bigint) into hex - i, _ := new(big.Int).SetString(args.Key, 10) - hx = ethutil.Bytes2Hex(i.Bytes()) - } - rpclogger.Debugf("GetStateAt(%s, %s)\n", args.Address, hx) - *reply = map[string]string{args.Key: value.Str()} - return nil -} - -func (p *EthereumApi) GetStorageAt(args *GetStorageArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - - *reply = p.xeth.State().SafeGet(args.Address).Storage() - return nil -} - -func (p *EthereumApi) GetPeerCount(reply *interface{}) error { - *reply = p.xeth.PeerCount() - return nil -} - -func (p *EthereumApi) GetIsListening(reply *interface{}) error { - *reply = p.xeth.IsListening() - return nil -} - -func (p *EthereumApi) GetCoinbase(reply *interface{}) error { - *reply = p.xeth.Coinbase() - return nil -} - -func (p *EthereumApi) Accounts(reply *interface{}) error { - *reply = p.xeth.Accounts() - return nil -} - -func (p *EthereumApi) GetIsMining(reply *interface{}) error { - *reply = p.xeth.IsMining() - return nil -} - -func (p *EthereumApi) SetMining(shouldmine bool, reply *interface{}) error { - *reply = p.xeth.SetMining(shouldmine) - return nil -} - -func (p *EthereumApi) BlockNumber(reply *interface{}) error { - *reply = p.xeth.Backend().ChainManager().CurrentBlock().Number() - return nil -} - -func (p *EthereumApi) GetTxCountAt(args *GetTxCountArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - *reply = p.xeth.TxCountAt(args.Address) - return nil -} - -func (p *EthereumApi) GetBalanceAt(args *GetBalanceArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - state := p.xeth.State().SafeGet(args.Address) - *reply = toHex(state.Balance().Bytes()) - return nil -} - -func (p *EthereumApi) GetCodeAt(args *GetCodeAtArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - *reply = p.xeth.CodeAt(args.Address) - return nil -} - -func (p *EthereumApi) GetCompilers(reply *interface{}) error { - c := []string{"serpent"} - *reply = c - return nil -} - -func (p *EthereumApi) CompileSerpent(script string, reply *interface{}) error { - res, err := ethutil.Compile(script, false) - if err != nil { - return err - } - *reply = res - return nil -} - -func (p *EthereumApi) Sha3(args *Sha3Args, reply *interface{}) error { - *reply = toHex(crypto.Sha3(fromHex(args.Data))) - return nil -} - -func (p *EthereumApi) DbPut(args *DbArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - - p.db.Put([]byte(args.Database+args.Key), []byte(args.Value)) - *reply = true - return nil -} - -func (p *EthereumApi) DbGet(args *DbArgs, reply *interface{}) error { - err := args.requirements() - if err != nil { - return err - } - - res, _ := p.db.Get([]byte(args.Database + args.Key)) - *reply = string(res) - return nil -} - -func (p *EthereumApi) NewWhisperIdentity(reply *interface{}) error { - *reply = p.xeth.Whisper().NewIdentity() - return nil -} - -func (p *EthereumApi) NewWhisperFilter(args *xeth.Options, reply *interface{}) error { - var id int - args.Fn = func(msg xeth.WhisperMessage) { - p.messagesMut.Lock() - defer p.messagesMut.Unlock() - p.messages[id].add(msg) // = append(p.messages[id], msg) - } - id = p.xeth.Whisper().Watch(args) - p.messages[id] = &whisperFilter{timeout: time.Now()} - *reply = id - return nil -} - -func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error { - self.messagesMut.Lock() - defer self.messagesMut.Unlock() - - if self.messages[id] != nil { - *reply = self.messages[id].get() - } - - return nil -} - -func (p *EthereumApi) WhisperPost(args *WhisperMessageArgs, reply *interface{}) error { - err := p.xeth.Whisper().Post(args.Payload, args.To, args.From, args.Topic, args.Priority, args.Ttl) - if err != nil { - return err - } - - *reply = true - return nil -} - -func (p *EthereumApi) HasWhisperIdentity(args string, reply *interface{}) error { - *reply = p.xeth.Whisper().HasIdentity(args) - return nil -} - -func (p *EthereumApi) WhisperMessages(id int, reply *interface{}) error { - *reply = p.xeth.Whisper().Messages(id) - return nil -} - -func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error { - // Spec at https://github.com/ethereum/wiki/wiki/Generic-JSON-RPC - rpclogger.DebugDetailf("%T %s", req.Params, req.Params) - switch req.Method { - case "eth_coinbase": - return p.GetCoinbase(reply) - case "eth_listening": - return p.GetIsListening(reply) - case "eth_mining": - return p.GetIsMining(reply) - case "eth_setMining": - args, err := req.ToBoolArgs() - if err != nil { - return err - } - return p.SetMining(args, reply) - case "eth_peerCount": - return p.GetPeerCount(reply) - case "eth_number": - return p.BlockNumber(reply) - case "eth_accounts": - return p.Accounts(reply) - case "eth_countAt": - args, err := req.ToGetTxCountArgs() - if err != nil { - return err - } - return p.GetTxCountAt(args, reply) - case "eth_codeAt": - args, err := req.ToGetCodeAtArgs() - if err != nil { - return err - } - return p.GetCodeAt(args, reply) - case "eth_balanceAt": - args, err := req.ToGetBalanceArgs() - if err != nil { - return err - } - return p.GetBalanceAt(args, reply) - case "eth_stateAt": - args, err := req.ToGetStateArgs() - if err != nil { - return err - } - return p.GetStateAt(args, reply) - case "eth_storageAt": - args, err := req.ToStorageAtArgs() - if err != nil { - return err - } - return p.GetStorageAt(args, reply) - case "eth_blockByNumber", "eth_blockByHash": - args, err := req.ToGetBlockArgs() - if err != nil { - return err - } - return p.GetBlock(args, reply) - case "eth_transact": - args, err := req.ToNewTxArgs() - if err != nil { - return err - } - return p.Transact(args, reply) - case "eth_call": - args, err := req.ToNewTxArgs() - if err != nil { - return err - } - return p.Call(args, reply) - case "eth_newFilter": - args, err := req.ToFilterArgs() - if err != nil { - return err - } - return p.NewFilter(args, reply) - case "eth_newFilterString": - args, err := req.ToFilterStringArgs() - if err != nil { - return err - } - return p.NewFilterString(args, reply) - case "eth_uninstallFilter": - args, err := req.ToUninstallFilterArgs() - if err != nil { - return err - } - return p.UninstallFilter(args, reply) - case "eth_changed": - args, err := req.ToIdArgs() - if err != nil { - return err - } - return p.FilterChanged(args, reply) - case "eth_filterLogs": - args, err := req.ToIdArgs() - if err != nil { - return err - } - return p.Logs(args, reply) - case "eth_logs": - args, err := req.ToFilterArgs() - if err != nil { - return err - } - return p.AllLogs(args, reply) - case "eth_gasPrice": - *reply = toHex(defaultGasPrice.Bytes()) - return nil - case "eth_register": - args, err := req.ToRegisterArgs() - if err != nil { - return err - } - return p.Register(args, reply) - case "eth_unregister": - args, err := req.ToRegisterArgs() - if err != nil { - return err - } - return p.Unregister(args, reply) - case "eth_watchTx": - args, err := req.ToWatchTxArgs() - if err != nil { - return err - } - return p.WatchTx(args, reply) - case "eth_compilers": - return p.GetCompilers(reply) - case "eth_serpent": - args, err := req.ToCompileArgs() - if err != nil { - return err - } - return p.CompileSerpent(args, reply) - case "web3_sha3": - args, err := req.ToSha3Args() - if err != nil { - return err - } - return p.Sha3(args, reply) - case "db_put": - args, err := req.ToDbPutArgs() - if err != nil { - return err - } - return p.DbPut(args, reply) - case "db_get": - args, err := req.ToDbGetArgs() - if err != nil { - return err - } - return p.DbGet(args, reply) - case "shh_newIdentity": - return p.NewWhisperIdentity(reply) - case "shh_newFilter": - args, err := req.ToWhisperFilterArgs() - if err != nil { - return err - } - return p.NewWhisperFilter(args, reply) - case "shh_changed": - args, err := req.ToIdArgs() - if err != nil { - return err - } - return p.MessagesChanged(args, reply) - case "shh_post": - args, err := req.ToWhisperPostArgs() - if err != nil { - return err - } - return p.WhisperPost(args, reply) - case "shh_haveIdentity": - args, err := req.ToWhisperHasIdentityArgs() - if err != nil { - return err - } - return p.HasWhisperIdentity(args, reply) - case "shh_getMessages": - args, err := req.ToIdArgs() - if err != nil { - return err - } - return p.WhisperMessages(args, reply) - default: - return NewErrorWithMessage(errNotImplemented, req.Method) - } - - rpclogger.DebugDetailf("Reply: %T %s", reply, reply) - return nil -} - -var filterTickerTime = 15 * time.Second - -func (self *EthereumApi) start() { - timer := time.NewTicker(filterTickerTime) -done: - for { - select { - case <-timer.C: - self.logMut.Lock() - self.messagesMut.Lock() - for id, filter := range self.logs { - if time.Since(filter.timeout) > 20*time.Second { - self.filterManager.UninstallFilter(id) - delete(self.logs, id) - } - } - - for id, filter := range self.messages { - if time.Since(filter.timeout) > 20*time.Second { - self.xeth.Whisper().Unwatch(id) - delete(self.messages, id) - } - } - self.logMut.Unlock() - self.messagesMut.Unlock() - case <-self.quit: - break done - } - } -} - -func (self *EthereumApi) stop() { - close(self.quit) -} diff --git a/rpc/packages_test.go b/rpc/packages_test.go deleted file mode 100644 index a9fc16cd3..000000000 --- a/rpc/packages_test.go +++ /dev/null @@ -1,38 +0,0 @@ -package rpc - -import ( - "sync" - "testing" - "time" -) - -func TestFilterClose(t *testing.T) { - t.Skip() - api := &EthereumApi{ - logs: make(map[int]*logFilter), - messages: make(map[int]*whisperFilter), - quit: make(chan struct{}), - } - - filterTickerTime = 1 - api.logs[0] = &logFilter{} - api.messages[0] = &whisperFilter{} - var wg sync.WaitGroup - wg.Add(1) - go api.start() - go func() { - select { - case <-time.After(500 * time.Millisecond): - api.stop() - wg.Done() - } - }() - wg.Wait() - if len(api.logs) != 0 { - t.Error("expected logs to be empty") - } - - if len(api.messages) != 0 { - t.Error("expected messages to be empty") - } -} -- cgit From c8e9abff53d869f74211e157abcb2827d181e106 Mon Sep 17 00:00:00 2001 From: Taylor Gerring Date: Tue, 24 Feb 2015 19:22:08 +0100 Subject: Remove errParseRequest --- rpc/http/server.go | 2 +- rpc/messages.go | 1 - rpc/ws/server.go | 2 +- 3 files changed, 2 insertions(+), 3 deletions(-) diff --git a/rpc/http/server.go b/rpc/http/server.go index dd6ba68e3..fa66eed48 100644 --- a/rpc/http/server.go +++ b/rpc/http/server.go @@ -92,7 +92,7 @@ func (s *RpcHttpServer) apiHandler(api *rpc.EthereumApi) http.Handler { reqParsed, reqerr := JSON.ParseRequestBody(req) if reqerr != nil { - jsonerr := &rpc.RpcErrorObject{-32700, rpc.ErrorParseRequest} + jsonerr := &rpc.RpcErrorObject{-32700, "Error: Could not parse request"} JSON.Send(w, &rpc.RpcErrorResponse{JsonRpc: jsonrpcver, ID: nil, Error: jsonerr}) return } diff --git a/rpc/messages.go b/rpc/messages.go index 18fbec931..044f07545 100644 --- a/rpc/messages.go +++ b/rpc/messages.go @@ -29,7 +29,6 @@ var ( errArguments = errors.New("Error: Insufficient arguments") errNotImplemented = errors.New("Error: Method not implemented") errUnknown = errors.New("Error: Unknown error") - errParseRequest = errors.New("Error: Could not parse request") errDecodeArgs = errors.New("Error: Could not decode arguments") ) diff --git a/rpc/ws/server.go b/rpc/ws/server.go index b8cc2fa6b..2c2988f5d 100644 --- a/rpc/ws/server.go +++ b/rpc/ws/server.go @@ -99,7 +99,7 @@ func sockHandler(api *rpc.EthereumApi) websocket.Handler { // reqParsed, reqerr := JSON.ParseRequestBody(conn.Request()) if err := websocket.JSON.Receive(conn, &reqParsed); err != nil { - jsonerr := &rpc.RpcErrorObject{-32700, rpc.ErrorParseRequest} + jsonerr := &rpc.RpcErrorObject{-32700, "Error: Could not parse request"} JSON.Send(conn, &rpc.RpcErrorResponse{JsonRpc: jsonrpcver, ID: nil, Error: jsonerr}) continue } -- cgit From 2e3a6e2559d02ba90957eaf333e571dec935e00a Mon Sep 17 00:00:00 2001 From: Taylor Gerring Date: Tue, 24 Feb 2015 19:54:18 +0100 Subject: Consolidate related items --- rpc/api.go | 73 +++++++++++++++++++++++++++++++------------------------------- 1 file changed, 36 insertions(+), 37 deletions(-) diff --git a/rpc/api.go b/rpc/api.go index 813fef949..21c85bbcc 100644 --- a/rpc/api.go +++ b/rpc/api.go @@ -25,8 +25,9 @@ import ( ) var ( - defaultGasPrice = big.NewInt(10000000000000) - defaultGas = big.NewInt(10000) + defaultGasPrice = big.NewInt(10000000000000) + defaultGas = big.NewInt(10000) + filterTickerTime = 15 * time.Second ) type EthereumApi struct { @@ -62,6 +63,39 @@ func NewEthereumApi(eth *xeth.XEth) *EthereumApi { return api } +func (self *EthereumApi) start() { + timer := time.NewTicker(filterTickerTime) +done: + for { + select { + case <-timer.C: + self.logMut.Lock() + self.messagesMut.Lock() + for id, filter := range self.logs { + if time.Since(filter.timeout) > 20*time.Second { + self.filterManager.UninstallFilter(id) + delete(self.logs, id) + } + } + + for id, filter := range self.messages { + if time.Since(filter.timeout) > 20*time.Second { + self.xeth.Whisper().Unwatch(id) + delete(self.messages, id) + } + } + self.logMut.Unlock() + self.messagesMut.Unlock() + case <-self.quit: + break done + } + } +} + +func (self *EthereumApi) stop() { + close(self.quit) +} + func (self *EthereumApi) Register(args string, reply *interface{}) error { self.regmut.Lock() defer self.regmut.Unlock() @@ -600,38 +634,3 @@ func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error rpclogger.DebugDetailf("Reply: %T %s", reply, reply) return nil } - -var filterTickerTime = 15 * time.Second - -func (self *EthereumApi) start() { - timer := time.NewTicker(filterTickerTime) -done: - for { - select { - case <-timer.C: - self.logMut.Lock() - self.messagesMut.Lock() - for id, filter := range self.logs { - if time.Since(filter.timeout) > 20*time.Second { - self.filterManager.UninstallFilter(id) - delete(self.logs, id) - } - } - - for id, filter := range self.messages { - if time.Since(filter.timeout) > 20*time.Second { - self.xeth.Whisper().Unwatch(id) - delete(self.messages, id) - } - } - self.logMut.Unlock() - self.messagesMut.Unlock() - case <-self.quit: - break done - } - } -} - -func (self *EthereumApi) stop() { - close(self.quit) -} -- cgit