diff options
Diffstat (limited to 'rpc')
-rw-r--r-- | rpc/args.go | 252 | ||||
-rw-r--r-- | rpc/http/server.go | 114 | ||||
-rw-r--r-- | rpc/json.go | 20 | ||||
-rw-r--r-- | rpc/message.go | 346 | ||||
-rw-r--r-- | rpc/packages.go | 534 | ||||
-rw-r--r-- | rpc/server.go | 67 | ||||
-rw-r--r-- | rpc/util.go | 102 | ||||
-rw-r--r-- | rpc/writer.go | 75 | ||||
-rw-r--r-- | rpc/ws/server.go | 121 |
9 files changed, 1280 insertions, 351 deletions
diff --git a/rpc/args.go b/rpc/args.go new file mode 100644 index 000000000..84b076d4a --- /dev/null +++ b/rpc/args.go @@ -0,0 +1,252 @@ +package rpc + +import "encoding/json" +import "github.com/ethereum/go-ethereum/core" + +type GetBlockArgs struct { + BlockNumber int32 + Hash string +} + +func (obj *GetBlockArgs) UnmarshalJSON(b []byte) (err error) { + argint, argstr := int32(0), "" + if err = json.Unmarshal(b, &argint); err == nil { + obj.BlockNumber = argint + return + } + if err = json.Unmarshal(b, &argstr); err == nil { + obj.Hash = argstr + return + } + return NewErrorResponse(ErrorDecodeArgs) +} + +func (obj *GetBlockArgs) requirements() error { + if obj.BlockNumber == 0 && obj.Hash == "" { + return NewErrorResponse("GetBlock requires either a block 'number' or a block 'hash' as argument") + } + return nil +} + +type NewTxArgs struct { + From string `json:"from"` + To string `json:"to"` + Value string `json:"value"` + Gas string `json:"gas"` + GasPrice string `json:"gasPrice"` + Data string `json:"data"` +} + +type PushTxArgs struct { + Tx string `json:"tx"` +} + +func (obj *PushTxArgs) UnmarshalJSON(b []byte) (err error) { + arg0 := "" + if err = json.Unmarshal(b, arg0); err == nil { + obj.Tx = arg0 + return + } + return NewErrorResponse(ErrorDecodeArgs) +} + +func (a *PushTxArgs) requirementsPushTx() error { + if a.Tx == "" { + return NewErrorResponse("PushTx requires a 'tx' as argument") + } + return nil +} + +type GetStorageArgs struct { + Address string +} + +func (obj *GetStorageArgs) UnmarshalJSON(b []byte) (err error) { + if err = json.Unmarshal(b, &obj.Address); err != nil { + return NewErrorResponse(ErrorDecodeArgs) + } + return +} + +func (a *GetStorageArgs) requirements() error { + if len(a.Address) == 0 { + return NewErrorResponse("GetStorageAt requires an 'address' value as argument") + } + return nil +} + +type GetStateArgs struct { + Address string + Key string +} + +func (obj *GetStateArgs) UnmarshalJSON(b []byte) (err error) { + arg0 := "" + if err = json.Unmarshal(b, arg0); err == nil { + obj.Address = arg0 + return + } + return NewErrorResponse(ErrorDecodeArgs) +} + +func (a *GetStateArgs) requirements() error { + if a.Address == "" { + return NewErrorResponse("GetStorageAt requires an 'address' value as argument") + } + if a.Key == "" { + return NewErrorResponse("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") +} + +func (a *GetTxCountArgs) requirements() error { + if a.Address == "" { + return NewErrorResponse("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 +} + +func (obj *GetBalanceArgs) 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") +} + +func (a *GetBalanceArgs) requirements() error { + if a.Address == "" { + return NewErrorResponse("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 +} + +func (obj *GetCodeAtArgs) UnmarshalJSON(b []byte) (err error) { + arg0 := "" + if err = json.Unmarshal(b, &arg0); err == nil { + obj.Address = arg0 + return + } + return NewErrorResponse(ErrorDecodeArgs) +} + +func (a *GetCodeAtArgs) requirements() error { + if a.Address == "" { + return NewErrorResponse("GetCodeAt requires an 'address' value as argument") + } + return nil +} + +type Sha3Args struct { + Data string +} + +func (obj *Sha3Args) UnmarshalJSON(b []byte) (err error) { + if err = json.Unmarshal(b, &obj.Data); err != nil { + return NewErrorResponse(ErrorDecodeArgs) + } + return +} + +type FilterOptions struct { + Earliest int64 + Latest int64 + Address string + Topic []string + Skip int + Max int +} + +func toFilterOptions(options *FilterOptions) core.FilterOptions { + var opts core.FilterOptions + opts.Earliest = options.Earliest + opts.Latest = options.Latest + opts.Address = fromHex(options.Address) + opts.Topics = make([][]byte, len(options.Topic)) + for i, topic := range options.Topic { + opts.Topics[i] = fromHex(topic) + } + + return opts +} + +type FilterChangedArgs struct { + n int +} + +type DbArgs struct { + Database string + Key string + Value string +} + +func (a *DbArgs) requirements() error { + if len(a.Database) == 0 { + return NewErrorResponse("DbPutArgs requires an 'Database' value as argument") + } + if len(a.Key) == 0 { + return NewErrorResponse("DbPutArgs requires an 'Key' value as argument") + } + return nil +} + +type WhisperMessageArgs struct { + Payload string + To string + From string + Topics []string + Priority uint32 + Ttl uint32 +} diff --git a/rpc/http/server.go b/rpc/http/server.go new file mode 100644 index 000000000..10c8fa813 --- /dev/null +++ b/rpc/http/server.go @@ -0,0 +1,114 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ +package rpchttp + +import ( + "fmt" + "net" + "net/http" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/xeth" +) + +var rpchttplogger = logger.NewLogger("RPC-HTTP") +var JSON rpc.JsonWrapper + +func NewRpcHttpServer(pipe *xeth.XEth, port int) (*RpcHttpServer, error) { + sport := fmt.Sprintf(":%d", port) + l, err := net.Listen("tcp", sport) + if err != nil { + return nil, err + } + + return &RpcHttpServer{ + listener: l, + quit: make(chan bool), + pipe: pipe, + port: port, + }, nil +} + +type RpcHttpServer struct { + quit chan bool + listener net.Listener + pipe *xeth.XEth + port int +} + +func (s *RpcHttpServer) exitHandler() { +out: + for { + select { + case <-s.quit: + s.listener.Close() + break out + } + } + + rpchttplogger.Infoln("Shutdown RPC-HTTP server") +} + +func (s *RpcHttpServer) Stop() { + close(s.quit) +} + +func (s *RpcHttpServer) Start() { + rpchttplogger.Infof("Starting RPC-HTTP server on port %d", s.port) + go s.exitHandler() + + api := rpc.NewEthereumApi(s.pipe) + h := s.apiHandler(api) + http.Handle("/", h) + + err := http.Serve(s.listener, nil) + // FIX Complains on shutdown due to listner already being closed + if err != nil { + rpchttplogger.Errorln("Error on RPC-HTTP interface:", err) + } +} + +func (s *RpcHttpServer) apiHandler(api *rpc.EthereumApi) http.Handler { + var jsonrpcver string = "2.0" + fn := func(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + + rpchttplogger.DebugDetailln("Handling request") + + reqParsed, reqerr := JSON.ParseRequestBody(req) + if reqerr != nil { + jsonerr := &rpc.RpcErrorObject{-32700, rpc.ErrorParseRequest} + JSON.Send(w, &rpc.RpcErrorResponse{JsonRpc: jsonrpcver, ID: nil, Error: jsonerr}) + return + } + + var response interface{} + reserr := api.GetRequestReply(&reqParsed, &response) + if reserr != nil { + rpchttplogger.Warnln(reserr) + jsonerr := &rpc.RpcErrorObject{-32603, reserr.Error()} + JSON.Send(w, &rpc.RpcErrorResponse{JsonRpc: jsonrpcver, ID: reqParsed.ID, Error: jsonerr}) + return + } + + rpchttplogger.DebugDetailf("Generated response: %T %s", response, response) + JSON.Send(w, &rpc.RpcSuccessResponse{JsonRpc: jsonrpcver, ID: reqParsed.ID, Result: response}) + } + + return http.HandlerFunc(fn) +} diff --git a/rpc/json.go b/rpc/json.go deleted file mode 100644 index e467f9a34..000000000 --- a/rpc/json.go +++ /dev/null @@ -1,20 +0,0 @@ -package rpc - -import ( - "encoding/json" - "io" -) - -type jsonWrapper struct{} - -func (self jsonWrapper) Send(writer io.Writer, v interface{}) (n int, err error) { - var payload []byte - payload, err = json.Marshal(v) - if err != nil { - return 0, err - } - - return writer.Write(payload) -} - -var JSON jsonWrapper diff --git a/rpc/message.go b/rpc/message.go index c8103eded..7983e003d 100644 --- a/rpc/message.go +++ b/rpc/message.go @@ -1,14 +1,344 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ package rpc -import "github.com/ethereum/go-ethereum/ethutil" +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + + "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" +) + +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 NewErrorResponse(msg string) error { + return errors.New(msg) +} + +func NewErrorResponseWithError(msg string, err error) error { + return fmt.Errorf("%s: %v", msg, err) +} + +func (req *RpcRequest) ToSha3Args() (*Sha3Args, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } -type Message struct { - Call string `json:"call"` - Args []interface{} `json:"args"` - Id int `json:"_id"` - Data interface{} `json:"data"` + args := new(Sha3Args) + r := bytes.NewReader(req.Params[0]) + if err := json.NewDecoder(r).Decode(args); err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil } -func (self *Message) Arguments() *ethutil.Value { - return ethutil.NewValue(self.Args) +func (req *RpcRequest) ToGetBlockArgs() (*GetBlockArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(GetBlockArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToNewTxArgs() (*NewTxArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(NewTxArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToPushTxArgs() (*PushTxArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(PushTxArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToGetStateArgs() (*GetStateArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + 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, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToStorageAtArgs() (*GetStorageArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(GetStorageArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToGetTxCountArgs() (*GetTxCountArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(GetTxCountArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToGetBalanceArgs() (*GetBalanceArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(GetBalanceArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToGetCodeAtArgs() (*GetCodeAtArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(GetCodeAtArgs) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + args := new(FilterOptions) + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(args) + if err != nil { + return nil, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToFilterStringArgs() (string, error) { + if len(req.Params) < 1 { + return "", NewErrorResponse(ErrorArguments) + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return "", NewErrorResponse(ErrorDecodeArgs) + } + + rpclogger.DebugDetailf("%T %v", args, args) + return args, nil +} + +func (req *RpcRequest) ToFilterChangedArgs() (int, error) { + if len(req.Params) < 1 { + return 0, NewErrorResponse(ErrorArguments) + } + + var id int + r := bytes.NewReader(req.Params[0]) + err := json.NewDecoder(r).Decode(&id) + if err != nil { + return 0, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", id, id) + return id, nil +} + +func (req *RpcRequest) ToDbPutArgs() (*DbArgs, error) { + if len(req.Params) < 3 { + return nil, NewErrorResponse(ErrorArguments) + } + + var args DbArgs + err := json.Unmarshal(req.Params[0], &args.Database) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + err = json.Unmarshal(req.Params[1], &args.Key) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + err = json.Unmarshal(req.Params[2], &args.Value) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + rpclogger.DebugDetailf("%T %v", args, args) + return &args, nil +} + +func (req *RpcRequest) ToDbGetArgs() (*DbArgs, error) { + if len(req.Params) < 2 { + return nil, NewErrorResponse(ErrorArguments) + } + + var args DbArgs + err := json.Unmarshal(req.Params[0], &args.Database) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + + err = json.Unmarshal(req.Params[1], &args.Key) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + 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) + } + + var args xeth.Options + err := json.Unmarshal(req.Params[0], &args) + if err != nil { + return nil, NewErrorResponseWithError(ErrorDecodeArgs, err) + } + rpclogger.DebugDetailf("%T %v", args, args) + return &args, nil +} + +func (req *RpcRequest) ToWhisperIdArgs() (int, error) { + if len(req.Params) < 1 { + return 0, NewErrorResponse(ErrorArguments) + } + + var id int + err := json.Unmarshal(req.Params[0], &id) + if err != nil { + return 0, NewErrorResponse(ErrorDecodeArgs) + } + rpclogger.DebugDetailf("%T %v", id, id) + return id, nil +} + +func (req *RpcRequest) ToWhisperPostArgs() (*WhisperMessageArgs, error) { + if len(req.Params) < 1 { + return nil, NewErrorResponse(ErrorArguments) + } + + var args WhisperMessageArgs + err := json.Unmarshal(req.Params[0], &args) + 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) + } + + var args string + err := json.Unmarshal(req.Params[0], &args) + 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 5535e6e79..ac3127356 100644 --- a/rpc/packages.go +++ b/rpc/packages.go @@ -1,205 +1,188 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ +/* + +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 ( - "encoding/json" - "errors" + "fmt" "math/big" "strings" + "sync" + "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" ) +const ( + defaultGasPrice = "10000000000000" + defaultGas = "10000" +) + type EthereumApi struct { - pipe *xeth.JSXEth -} + xeth *xeth.XEth + filterManager *filter.FilterManager -type JsonArgs interface { - requirements() error -} + logMut sync.RWMutex + logs map[int]state.Logs -type BlockResponse struct { - JsonResponse -} -type GetBlockArgs struct { - BlockNumber int - Hash string -} + messagesMut sync.RWMutex + messages map[int][]xeth.WhisperMessage -type ErrorResponse struct { - Error bool `json:"error"` - ErrorText string `json:"errorText"` + db ethutil.Database } -type JsonResponse interface { -} +func NewEthereumApi(eth *xeth.XEth) *EthereumApi { + db, _ := ethdb.NewLDBDatabase("dapps") + api := &EthereumApi{ + xeth: eth, + filterManager: filter.NewFilterManager(eth.Backend().EventMux()), + logs: make(map[int]state.Logs), + messages: make(map[int][]xeth.WhisperMessage), + db: db, + } + go api.filterManager.Start() -type SuccessRes struct { - Error bool `json:"error"` - Result JsonResponse `json:"result"` + return api } -func NewSuccessRes(object JsonResponse) string { - e := SuccessRes{Error: false, Result: object} - res, err := json.Marshal(e) - if err != nil { - // This should never happen - panic("Creating json error response failed, help") - } - success := string(res) - return success -} +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() -func NewErrorResponse(msg string) error { - e := ErrorResponse{Error: true, ErrorText: msg} - res, err := json.Marshal(e) - if err != nil { - // This should never happen - panic("Creating json error response failed, help") + self.logs[id] = append(self.logs[id], logs...) } - newErr := errors.New(string(res)) - return newErr -} + id = self.filterManager.InstallFilter(filter) + *reply = id -func (b *GetBlockArgs) requirements() error { - if b.BlockNumber == 0 && b.Hash == "" { - return NewErrorResponse("GetBlock requires either a block 'number' or a block 'hash' as argument") - } return nil } -func (p *EthereumApi) GetBlock(args *GetBlockArgs, reply *string) error { - err := args.requirements() - if err != nil { - return err +func (self *EthereumApi) NewFilterString(args string, reply *interface{}) error { + var id int + filter := core.NewFilter(self.xeth.Backend()) + + callback := func(block *types.Block) { + self.logs[id] = append(self.logs[id], &state.StateLog{}) } + if args == "pending" { + filter.PendingCallback = callback + } else if args == "chain" { + filter.BlockCallback = callback + } + + id = self.filterManager.InstallFilter(filter) + *reply = id - block := p.pipe.BlockByHash(args.Hash) - *reply = NewSuccessRes(block) return nil } -type NewTxArgs struct { - Sec string - Recipient string - Value string - Gas string - GasPrice string - Init string - Body string -} -type TxResponse struct { - Hash string -} +func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error { + self.logMut.RLock() + defer self.logMut.RUnlock() + + *reply = toLogs(self.logs[id]) + + self.logs[id] = nil // empty the logs -func (a *NewTxArgs) requirements() error { - if a.Recipient == "" { - return NewErrorResponse("Transact requires a 'recipient' address as argument") - } - if a.Value == "" { - return NewErrorResponse("Transact requires a 'value' as argument") - } - if a.Gas == "" { - return NewErrorResponse("Transact requires a 'gas' value as argument") - } - if a.GasPrice == "" { - return NewErrorResponse("Transact requires a 'gasprice' value as argument") - } return nil } -func (a *NewTxArgs) requirementsContract() error { - if a.Value == "" { - return NewErrorResponse("Create requires a 'value' as argument") - } - if a.Gas == "" { - return NewErrorResponse("Create requires a 'gas' value as argument") - } - if a.GasPrice == "" { - return NewErrorResponse("Create requires a 'gasprice' value as argument") - } - if a.Body == "" { - return NewErrorResponse("Create requires a 'body' value as argument") - } +func (self *EthereumApi) Logs(id int, reply *interface{}) error { + filter := self.filterManager.GetFilter(id) + *reply = toLogs(filter.Find()) + return nil } -func (p *EthereumApi) Transact(args *NewTxArgs, reply *string) error { +func (p *EthereumApi) GetBlock(args *GetBlockArgs, reply *interface{}) error { err := args.requirements() if err != nil { return err } - result, _ := p.pipe.Transact(p.pipe.Key().PrivateKey, args.Recipient, args.Value, args.Gas, args.GasPrice, args.Body) - *reply = NewSuccessRes(result) - return nil -} -func (p *EthereumApi) Create(args *NewTxArgs, reply *string) error { - err := args.requirementsContract() - if err != nil { - return err + if args.BlockNumber > 0 { + *reply = p.xeth.BlockByNumber(args.BlockNumber) + } else { + *reply = p.xeth.BlockByHash(args.Hash) } - - result, _ := p.pipe.Transact(p.pipe.Key().PrivateKey, "", args.Value, args.Gas, args.GasPrice, args.Body) - *reply = NewSuccessRes(result) return nil } -type PushTxArgs struct { - Tx string -} +func (p *EthereumApi) Transact(args *NewTxArgs, reply *interface{}) error { + if len(args.Gas) == 0 { + args.Gas = defaultGas + } -func (a *PushTxArgs) requirementsPushTx() error { - if a.Tx == "" { - return NewErrorResponse("PushTx requires a 'tx' as argument") + if len(args.GasPrice) == 0 { + args.GasPrice = defaultGasPrice } + + result, _ := p.xeth.Transact( /* TODO specify account */ args.To, args.Value, args.Gas, args.GasPrice, args.Data) + *reply = result return nil } -func (p *EthereumApi) PushTx(args *PushTxArgs, reply *string) error { - err := args.requirementsPushTx() +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 } - result, _ := p.pipe.PushTx(args.Tx) - *reply = NewSuccessRes(result) - return nil -} -func (p *EthereumApi) GetKey(args interface{}, reply *string) error { - *reply = NewSuccessRes(p.pipe.Key()) + *reply = result return nil } -type GetStorageArgs struct { - Address string - Key string -} - -func (a *GetStorageArgs) requirements() error { - if a.Address == "" { - return NewErrorResponse("GetStorageAt requires an 'address' value as argument") - } - if a.Key == "" { - return NewErrorResponse("GetStorageAt requires an 'key' value as argument") +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 } -type GetStorageAtRes struct { - Key string `json:"key"` - Value string `json:"value"` - Address string `json:"address"` -} - -func (p *EthereumApi) GetStorageAt(args *GetStorageArgs, reply *string) error { +func (p *EthereumApi) GetStateAt(args *GetStateArgs, reply *interface{}) error { err := args.requirements() if err != nil { return err } - state := p.pipe.World().SafeGet(ethutil.Hex2Bytes(args.Address)) + 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:]) @@ -208,104 +191,293 @@ func (p *EthereumApi) GetStorageAt(args *GetStorageArgs, reply *string) error { i, _ := new(big.Int).SetString(args.Key, 10) hx = ethutil.Bytes2Hex(i.Bytes()) } - jsonlogger.Debugf("GetStorageAt(%s, %s)\n", args.Address, hx) - value := state.Storage(ethutil.Hex2Bytes(hx)) - *reply = NewSuccessRes(GetStorageAtRes{Address: args.Address, Key: args.Key, Value: value.Str()}) + rpclogger.Debugf("GetStateAt(%s, %s)\n", args.Address, hx) + *reply = map[string]string{args.Key: value.Str()} return nil } -type GetTxCountArgs struct { - Address string `json:"address"` +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 } -type GetTxCountRes struct { - Nonce int `json:"nonce"` + +func (p *EthereumApi) GetPeerCount(reply *interface{}) error { + *reply = p.xeth.PeerCount() + return nil } -func (a *GetTxCountArgs) requirements() error { - if a.Address == "" { - return NewErrorResponse("GetTxCountAt requires an 'address' value as argument") - } +func (p *EthereumApi) GetIsListening(reply *interface{}) error { + *reply = p.xeth.IsListening() return nil } -type GetPeerCountRes struct { - PeerCount int `json:"peerCount"` +func (p *EthereumApi) GetCoinbase(reply *interface{}) error { + *reply = p.xeth.Coinbase() + return nil } -func (p *EthereumApi) GetPeerCount(args *interface{}, reply *string) error { - *reply = NewSuccessRes(GetPeerCountRes{PeerCount: p.pipe.PeerCount()}) +func (p *EthereumApi) Accounts(reply *interface{}) error { + *reply = p.xeth.Accounts() return nil } -type GetListeningRes struct { - IsListening bool `json:"isListening"` +func (p *EthereumApi) GetIsMining(reply *interface{}) error { + *reply = p.xeth.IsMining() + return nil } -func (p *EthereumApi) GetIsListening(args *interface{}, reply *string) error { - *reply = NewSuccessRes(GetListeningRes{IsListening: p.pipe.IsListening()}) +func (p *EthereumApi) BlockNumber(reply *interface{}) error { + *reply = p.xeth.Backend().ChainManager().CurrentBlock().Number() return nil } -type GetCoinbaseRes struct { - Coinbase string `json:"coinbase"` +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) GetCoinbase(args *interface{}, reply *string) error { - *reply = NewSuccessRes(GetCoinbaseRes{Coinbase: p.pipe.CoinBase()}) +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 } -type GetMiningRes struct { - IsMining bool `json:"isMining"` +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) GetIsMining(args *interface{}, reply *string) error { - *reply = NewSuccessRes(GetMiningRes{IsMining: p.pipe.IsMining()}) +func (p *EthereumApi) Sha3(args *Sha3Args, reply *interface{}) error { + *reply = toHex(crypto.Sha3(fromHex(args.Data))) return nil } -func (p *EthereumApi) GetTxCountAt(args *GetTxCountArgs, reply *string) error { +func (p *EthereumApi) DbPut(args *DbArgs, reply *interface{}) error { err := args.requirements() if err != nil { return err } - state := p.pipe.TxCountAt(args.Address) - *reply = NewSuccessRes(GetTxCountRes{Nonce: state}) + + 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 } -type GetBalanceArgs struct { - Address string +func (p *EthereumApi) NewWhisperIdentity(reply *interface{}) error { + *reply = p.xeth.Whisper().NewIdentity() + return nil } -func (a *GetBalanceArgs) requirements() error { - if a.Address == "" { - return NewErrorResponse("GetBalanceAt requires an 'address' value as argument") +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] = append(p.messages[id], msg) } + id = p.xeth.Whisper().Watch(args) + *reply = id return nil } -type BalanceRes struct { - Balance string `json:"balance"` - Address string `json:"address"` +func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error { + self.messagesMut.RLock() + defer self.messagesMut.RUnlock() + + *reply = self.messages[id] + + self.messages[id] = nil // empty the messages + + return nil } -func (p *EthereumApi) GetBalanceAt(args *GetBalanceArgs, reply *string) error { - err := args.requirements() +func (p *EthereumApi) WhisperPost(args *WhisperMessageArgs, reply *interface{}) error { + err := p.xeth.Whisper().Post(args.Payload, args.To, args.From, args.Topics, args.Priority, args.Ttl) if err != nil { return err } - state := p.pipe.World().SafeGet(ethutil.Hex2Bytes(args.Address)) - *reply = NewSuccessRes(BalanceRes{Balance: state.Balance().String(), Address: args.Address}) + + *reply = true return nil } -type TestRes struct { - JsonResponse `json:"-"` - Answer int `json:"answer"` +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) Test(args *GetBlockArgs, reply *string) error { - *reply = NewSuccessRes(TestRes{Answer: 15}) +func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error { + // Spec at https://github.com/ethereum/wiki/wiki/Generic-ON-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_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_changed": + args, err := req.ToFilterChangedArgs() + if err != nil { + return err + } + return p.FilterChanged(args, reply) + case "eth_gasPrice": + *reply = defaultGasPrice + return nil + 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.ToWhisperIdArgs() + 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.ToWhisperIdArgs() + if err != nil { + return err + } + return p.WhisperMessages(args, reply) + default: + return NewErrorResponse(fmt.Sprintf("%v %s", ErrorNotImplemented, req.Method)) + } + + rpclogger.DebugDetailf("Reply: %T %s", reply, reply) return nil } diff --git a/rpc/server.go b/rpc/server.go deleted file mode 100644 index 983dc6c33..000000000 --- a/rpc/server.go +++ /dev/null @@ -1,67 +0,0 @@ -package rpc - -import ( - "fmt" - "net" - "net/rpc" - "net/rpc/jsonrpc" - - "github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/xeth" -) - -var jsonlogger = logger.NewLogger("JSON") - -type JsonRpcServer struct { - quit chan bool - listener net.Listener - pipe *xeth.JSXEth -} - -func (s *JsonRpcServer) exitHandler() { -out: - for { - select { - case <-s.quit: - s.listener.Close() - break out - } - } - - jsonlogger.Infoln("Shutdown JSON-RPC server") -} - -func (s *JsonRpcServer) Stop() { - close(s.quit) -} - -func (s *JsonRpcServer) Start() { - jsonlogger.Infoln("Starting JSON-RPC server") - go s.exitHandler() - rpc.Register(&EthereumApi{pipe: s.pipe}) - rpc.HandleHTTP() - - for { - conn, err := s.listener.Accept() - if err != nil { - jsonlogger.Infoln("Error starting JSON-RPC:", err) - break - } - jsonlogger.Debugln("Incoming request.") - go jsonrpc.ServeConn(conn) - } -} - -func NewJsonRpcServer(pipe *xeth.JSXEth, port int) (*JsonRpcServer, error) { - sport := fmt.Sprintf(":%d", port) - l, err := net.Listen("tcp", sport) - if err != nil { - return nil, err - } - - return &JsonRpcServer{ - listener: l, - quit: make(chan bool), - pipe: pipe, - }, nil -} diff --git a/rpc/util.go b/rpc/util.go new file mode 100644 index 000000000..679d83754 --- /dev/null +++ b/rpc/util.go @@ -0,0 +1,102 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ +package rpc + +import ( + "encoding/json" + "io" + "net/http" + + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/state" +) + +var rpclogger = logger.NewLogger("RPC") + +type JsonWrapper struct{} + +func (self JsonWrapper) Send(writer io.Writer, v interface{}) (n int, err error) { + var payload []byte + payload, err = json.Marshal(v) + if err != nil { + rpclogger.Fatalln("Error marshalling JSON", err) + return 0, err + } + rpclogger.DebugDetailf("Sending payload: %s", payload) + + return writer.Write(payload) +} + +func (self JsonWrapper) ParseRequestBody(req *http.Request) (RpcRequest, error) { + var reqParsed RpcRequest + + // Convert JSON to native types + d := json.NewDecoder(req.Body) + defer req.Body.Close() + err := d.Decode(&reqParsed) + + if err != nil { + rpclogger.Errorln("Error decoding JSON: ", err) + return reqParsed, err + } + + rpclogger.DebugDetailf("Parsed request: %s", reqParsed) + + return reqParsed, nil +} + +func toHex(b []byte) string { + return "0x" + ethutil.Bytes2Hex(b) +} +func fromHex(s string) []byte { + if len(s) > 1 { + if s[0:2] == "0x" { + s = s[2:] + } + return ethutil.Hex2Bytes(s) + } + return nil +} + +type RpcServer interface { + Start() + Stop() +} + +type Log struct { + Address string `json:"address"` + Topics []string `json:"topics"` + Data string `json:"data"` +} + +func toLogs(logs state.Logs) (ls []Log) { + ls = make([]Log, len(logs)) + + for i, log := range logs { + var l Log + l.Topics = make([]string, len(log.Topics())) + l.Address = toHex(log.Address()) + l.Data = toHex(log.Data()) + for j, topic := range log.Topics() { + l.Topics[j] = toHex(topic) + } + ls[i] = l + } + + return +} diff --git a/rpc/writer.go b/rpc/writer.go deleted file mode 100644 index 4cc39f577..000000000 --- a/rpc/writer.go +++ /dev/null @@ -1,75 +0,0 @@ -package rpc - -/* -func pack(id int, v ...interface{}) Message { - return Message{Data: v, Id: id} -} - -func WriteOn(msg *Message, writer io.Writer) { - //msg := &Message{Seed: seed, Data: data} - - switch msg.Call { - case "compile": - data := ethutil.NewValue(msg.Args) - bcode, err := ethutil.Compile(data.Get(0).Str(), false) - if err != nil { - JSON.Send(writer, pack(msg.Id, err.Error())) - } - - code := ethutil.Bytes2Hex(bcode) - - JSON.Send(writer, pack(msg.Id, code, nil)) - case "block": - args := msg.Arguments() - - block := pipe.BlockByNumber(int32(args.Get(0).Uint())) - - JSON.Send(writer, pack(msg.Id, block)) - case "transact": - if mp, ok := msg.Args[0].(map[string]interface{}); ok { - object := mapToTxParams(mp) - JSON.Send( - writer, - pack(msg.Id, args(pipe.Transact(object["from"], object["to"], object["value"], object["gas"], object["gasPrice"], object["data"]))), - ) - - } - case "coinbase": - JSON.Send(writer, pack(msg.Id, pipe.CoinBase(), msg.Seed)) - - case "listening": - JSON.Send(writer, pack(msg.Id, pipe.IsListening())) - - case "mining": - JSON.Send(writer, pack(msg.Id, pipe.IsMining())) - - case "peerCoint": - JSON.Send(writer, pack(msg.Id, pipe.PeerCount())) - - case "countAt": - args := msg.Arguments() - - JSON.Send(writer, pack(msg.Id, pipe.TxCountAt(args.Get(0).Str()))) - - case "codeAt": - args := msg.Arguments() - - JSON.Send(writer, pack(msg.Id, len(pipe.CodeAt(args.Get(0).Str())))) - - case "stateAt": - args := msg.Arguments() - - JSON.Send(writer, pack(msg.Id, pipe.StorageAt(args.Get(0).Str(), args.Get(1).Str()))) - - case "balanceAt": - args := msg.Arguments() - - JSON.Send(writer, pack(msg.Id, pipe.BalanceAt(args.Get(0).Str()))) - - case "newFilter": - case "newFilterString": - case "messages": - // TODO - } -} -*/ diff --git a/rpc/ws/server.go b/rpc/ws/server.go new file mode 100644 index 000000000..100713c10 --- /dev/null +++ b/rpc/ws/server.go @@ -0,0 +1,121 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ +package rpcws + +import ( + "fmt" + "net" + "net/http" + + "code.google.com/p/go.net/websocket" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/xeth" +) + +var wslogger = logger.NewLogger("RPC-WS") +var JSON rpc.JsonWrapper + +type WebSocketServer struct { + pipe *xeth.XEth + port int + doneCh chan bool + listener net.Listener +} + +func NewWebSocketServer(pipe *xeth.XEth, port int) (*WebSocketServer, error) { + sport := fmt.Sprintf(":%d", port) + l, err := net.Listen("tcp", sport) + if err != nil { + return nil, err + } + + return &WebSocketServer{ + pipe, + port, + make(chan bool), + l, + }, nil +} + +func (self *WebSocketServer) handlerLoop() { + for { + select { + case <-self.doneCh: + wslogger.Infoln("Shutdown RPC-WS server") + return + } + } +} + +func (self *WebSocketServer) Stop() { + close(self.doneCh) +} + +func (self *WebSocketServer) Start() { + wslogger.Infof("Starting RPC-WS server on port %d", self.port) + go self.handlerLoop() + + api := rpc.NewEthereumApi(self.pipe) + h := self.apiHandler(api) + http.Handle("/ws", h) + + err := http.Serve(self.listener, nil) + if err != nil { + wslogger.Errorln("Error on RPC-WS interface:", err) + } +} + +func (s *WebSocketServer) apiHandler(api *rpc.EthereumApi) http.Handler { + fn := func(w http.ResponseWriter, req *http.Request) { + h := sockHandler(api) + s := websocket.Server{Handler: h} + s.ServeHTTP(w, req) + } + + return http.HandlerFunc(fn) +} + +func sockHandler(api *rpc.EthereumApi) websocket.Handler { + var jsonrpcver string = "2.0" + fn := func(conn *websocket.Conn) { + for { + wslogger.Debugln("Handling connection") + var reqParsed rpc.RpcRequest + + // reqParsed, reqerr := JSON.ParseRequestBody(conn.Request()) + if err := websocket.JSON.Receive(conn, &reqParsed); err != nil { + jsonerr := &rpc.RpcErrorObject{-32700, rpc.ErrorParseRequest} + JSON.Send(conn, &rpc.RpcErrorResponse{JsonRpc: jsonrpcver, ID: nil, Error: jsonerr}) + continue + } + + var response interface{} + reserr := api.GetRequestReply(&reqParsed, &response) + if reserr != nil { + wslogger.Warnln(reserr) + jsonerr := &rpc.RpcErrorObject{-32603, reserr.Error()} + JSON.Send(conn, &rpc.RpcErrorResponse{JsonRpc: jsonrpcver, ID: reqParsed.ID, Error: jsonerr}) + continue + } + + wslogger.Debugf("Generated response: %T %s", response, response) + JSON.Send(conn, &rpc.RpcSuccessResponse{JsonRpc: jsonrpcver, ID: reqParsed.ID, Result: response}) + } + } + return websocket.Handler(fn) +} |