123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372 |
- // Copyright 2015 The go-ethereum Authors
- // This file is part of the go-ethereum library.
- //
- // The go-ethereum library is free software: you can redistribute it and/or modify
- // it under the terms of the GNU Lesser General Public License as published by
- // the Free Software Foundation, either version 3 of the License, or
- // (at your option) any later version.
- //
- // The go-ethereum library is distributed in the hope that it will be useful,
- // but WITHOUT ANY WARRANTY; without even the implied warranty of
- // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- // GNU Lesser General Public License for more details.
- //
- // You should have received a copy of the GNU Lesser General Public License
- // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
- package rpc
- import (
- "bytes"
- "encoding/json"
- "fmt"
- "io"
- "reflect"
- "strconv"
- "strings"
- "sync"
- "github.com/ethereum/go-ethereum/log"
- )
- const (
- jsonrpcVersion = "2.0"
- serviceMethodSeparator = "_"
- subscribeMethodSuffix = "_subscribe"
- unsubscribeMethodSuffix = "_unsubscribe"
- notificationMethodSuffix = "_subscription"
- )
- type jsonRequest struct {
- Method string `json:"method"`
- Version string `json:"jsonrpc"`
- Id json.RawMessage `json:"id,omitempty"`
- Payload json.RawMessage `json:"params,omitempty"`
- }
- type jsonSuccessResponse struct {
- Version string `json:"jsonrpc"`
- Id interface{} `json:"id,omitempty"`
- Result interface{} `json:"result"`
- }
- type jsonError struct {
- Code int `json:"code"`
- Message string `json:"message"`
- Data interface{} `json:"data,omitempty"`
- }
- type jsonErrResponse struct {
- Version string `json:"jsonrpc"`
- Id interface{} `json:"id,omitempty"`
- Error jsonError `json:"error"`
- }
- type jsonSubscription struct {
- Subscription string `json:"subscription"`
- Result interface{} `json:"result,omitempty"`
- }
- type jsonNotification struct {
- Version string `json:"jsonrpc"`
- Method string `json:"method"`
- Params jsonSubscription `json:"params"`
- }
- // jsonCodec reads and writes JSON-RPC messages to the underlying connection. It
- // also has support for parsing arguments and serializing (result) objects.
- type jsonCodec struct {
- closer sync.Once // close closed channel once
- closed chan interface{} // closed on Close
- decMu sync.Mutex // guards the decoder
- decode func(v interface{}) error // decoder to allow multiple transports
- encMu sync.Mutex // guards the encoder
- encode func(v interface{}) error // encoder to allow multiple transports
- rw io.ReadWriteCloser // connection
- }
- func (err *jsonError) Error() string {
- if err.Message == "" {
- return fmt.Sprintf("json-rpc error %d", err.Code)
- }
- return err.Message
- }
- func (err *jsonError) ErrorCode() int {
- return err.Code
- }
- // NewCodec creates a new RPC server codec with support for JSON-RPC 2.0 based
- // on explicitly given encoding and decoding methods.
- func NewCodec(rwc io.ReadWriteCloser, encode, decode func(v interface{}) error) ServerCodec {
- return &jsonCodec{
- closed: make(chan interface{}),
- encode: encode,
- decode: decode,
- rw: rwc,
- }
- }
- // NewJSONCodec creates a new RPC server codec with support for JSON-RPC 2.0.
- func NewJSONCodec(rwc io.ReadWriteCloser) ServerCodec {
- enc := json.NewEncoder(rwc)
- dec := json.NewDecoder(rwc)
- dec.UseNumber()
- return &jsonCodec{
- closed: make(chan interface{}),
- encode: enc.Encode,
- decode: dec.Decode,
- rw: rwc,
- }
- }
- // isBatch returns true when the first non-whitespace characters is '['
- func isBatch(msg json.RawMessage) bool {
- for _, c := range msg {
- // skip insignificant whitespace (http://www.ietf.org/rfc/rfc4627.txt)
- if c == 0x20 || c == 0x09 || c == 0x0a || c == 0x0d {
- continue
- }
- return c == '['
- }
- return false
- }
- // ReadRequestHeaders will read new requests without parsing the arguments. It will
- // return a collection of requests, an indication if these requests are in batch
- // form or an error when the incoming message could not be read/parsed.
- func (c *jsonCodec) ReadRequestHeaders() ([]rpcRequest, bool, Error) {
- c.decMu.Lock()
- defer c.decMu.Unlock()
- var incomingMsg json.RawMessage
- if err := c.decode(&incomingMsg); err != nil {
- return nil, false, &invalidRequestError{err.Error()}
- }
- if isBatch(incomingMsg) {
- return parseBatchRequest(incomingMsg)
- }
- return parseRequest(incomingMsg)
- }
- // checkReqId returns an error when the given reqId isn't valid for RPC method calls.
- // valid id's are strings, numbers or null
- func checkReqId(reqId json.RawMessage) error {
- if len(reqId) == 0 {
- return fmt.Errorf("missing request id")
- }
- if _, err := strconv.ParseFloat(string(reqId), 64); err == nil {
- return nil
- }
- var str string
- if err := json.Unmarshal(reqId, &str); err == nil {
- return nil
- }
- return fmt.Errorf("invalid request id")
- }
- // parseRequest will parse a single request from the given RawMessage. It will return
- // the parsed request, an indication if the request was a batch or an error when
- // the request could not be parsed.
- func parseRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, Error) {
- var in jsonRequest
- if err := json.Unmarshal(incomingMsg, &in); err != nil {
- return nil, false, &invalidMessageError{err.Error()}
- }
- if err := checkReqId(in.Id); err != nil {
- return nil, false, &invalidMessageError{err.Error()}
- }
- // subscribe are special, they will always use `subscribeMethod` as first param in the payload
- if strings.HasSuffix(in.Method, subscribeMethodSuffix) {
- reqs := []rpcRequest{{id: &in.Id, isPubSub: true}}
- if len(in.Payload) > 0 {
- // first param must be subscription name
- var subscribeMethod [1]string
- if err := json.Unmarshal(in.Payload, &subscribeMethod); err != nil {
- log.Debug(fmt.Sprintf("Unable to parse subscription method: %v\n", err))
- return nil, false, &invalidRequestError{"Unable to parse subscription request"}
- }
- reqs[0].service, reqs[0].method = strings.TrimSuffix(in.Method, subscribeMethodSuffix), subscribeMethod[0]
- reqs[0].params = in.Payload
- return reqs, false, nil
- }
- return nil, false, &invalidRequestError{"Unable to parse subscription request"}
- }
- if strings.HasSuffix(in.Method, unsubscribeMethodSuffix) {
- return []rpcRequest{{id: &in.Id, isPubSub: true,
- method: in.Method, params: in.Payload}}, false, nil
- }
- elems := strings.Split(in.Method, serviceMethodSeparator)
- if len(elems) != 2 {
- return nil, false, &methodNotFoundError{in.Method, ""}
- }
- // regular RPC call
- if len(in.Payload) == 0 {
- return []rpcRequest{{service: elems[0], method: elems[1], id: &in.Id}}, false, nil
- }
- return []rpcRequest{{service: elems[0], method: elems[1], id: &in.Id, params: in.Payload}}, false, nil
- }
- // parseBatchRequest will parse a batch request into a collection of requests from the given RawMessage, an indication
- // if the request was a batch or an error when the request could not be read.
- func parseBatchRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, Error) {
- var in []jsonRequest
- if err := json.Unmarshal(incomingMsg, &in); err != nil {
- return nil, false, &invalidMessageError{err.Error()}
- }
- requests := make([]rpcRequest, len(in))
- for i, r := range in {
- if err := checkReqId(r.Id); err != nil {
- return nil, false, &invalidMessageError{err.Error()}
- }
- id := &in[i].Id
- // subscribe are special, they will always use `subscriptionMethod` as first param in the payload
- if strings.HasSuffix(r.Method, subscribeMethodSuffix) {
- requests[i] = rpcRequest{id: id, isPubSub: true}
- if len(r.Payload) > 0 {
- // first param must be subscription name
- var subscribeMethod [1]string
- if err := json.Unmarshal(r.Payload, &subscribeMethod); err != nil {
- log.Debug(fmt.Sprintf("Unable to parse subscription method: %v\n", err))
- return nil, false, &invalidRequestError{"Unable to parse subscription request"}
- }
- requests[i].service, requests[i].method = strings.TrimSuffix(r.Method, subscribeMethodSuffix), subscribeMethod[0]
- requests[i].params = r.Payload
- continue
- }
- return nil, true, &invalidRequestError{"Unable to parse (un)subscribe request arguments"}
- }
- if strings.HasSuffix(r.Method, unsubscribeMethodSuffix) {
- requests[i] = rpcRequest{id: id, isPubSub: true, method: r.Method, params: r.Payload}
- continue
- }
- if len(r.Payload) == 0 {
- requests[i] = rpcRequest{id: id, params: nil}
- } else {
- requests[i] = rpcRequest{id: id, params: r.Payload}
- }
- if elem := strings.Split(r.Method, serviceMethodSeparator); len(elem) == 2 {
- requests[i].service, requests[i].method = elem[0], elem[1]
- } else {
- requests[i].err = &methodNotFoundError{r.Method, ""}
- }
- }
- return requests, true, nil
- }
- // ParseRequestArguments tries to parse the given params (json.RawMessage) with the given
- // types. It returns the parsed values or an error when the parsing failed.
- func (c *jsonCodec) ParseRequestArguments(argTypes []reflect.Type, params interface{}) ([]reflect.Value, Error) {
- if args, ok := params.(json.RawMessage); !ok {
- return nil, &invalidParamsError{"Invalid params supplied"}
- } else {
- return parsePositionalArguments(args, argTypes)
- }
- }
- // parsePositionalArguments tries to parse the given args to an array of values with the
- // given types. It returns the parsed values or an error when the args could not be
- // parsed. Missing optional arguments are returned as reflect.Zero values.
- func parsePositionalArguments(rawArgs json.RawMessage, types []reflect.Type) ([]reflect.Value, Error) {
- // Read beginning of the args array.
- dec := json.NewDecoder(bytes.NewReader(rawArgs))
- if tok, _ := dec.Token(); tok != json.Delim('[') {
- return nil, &invalidParamsError{"non-array args"}
- }
- // Read args.
- args := make([]reflect.Value, 0, len(types))
- for i := 0; dec.More(); i++ {
- if i >= len(types) {
- return nil, &invalidParamsError{fmt.Sprintf("too many arguments, want at most %d", len(types))}
- }
- argval := reflect.New(types[i])
- if err := dec.Decode(argval.Interface()); err != nil {
- return nil, &invalidParamsError{fmt.Sprintf("invalid argument %d: %v", i, err)}
- }
- if argval.IsNil() && types[i].Kind() != reflect.Ptr {
- return nil, &invalidParamsError{fmt.Sprintf("missing value for required argument %d", i)}
- }
- args = append(args, argval.Elem())
- }
- // Read end of args array.
- if _, err := dec.Token(); err != nil {
- return nil, &invalidParamsError{err.Error()}
- }
- // Set any missing args to nil.
- for i := len(args); i < len(types); i++ {
- if types[i].Kind() != reflect.Ptr {
- return nil, &invalidParamsError{fmt.Sprintf("missing value for required argument %d", i)}
- }
- args = append(args, reflect.Zero(types[i]))
- }
- return args, nil
- }
- // CreateResponse will create a JSON-RPC success response with the given id and reply as result.
- func (c *jsonCodec) CreateResponse(id interface{}, reply interface{}) interface{} {
- if isHexNum(reflect.TypeOf(reply)) {
- return &jsonSuccessResponse{Version: jsonrpcVersion, Id: id, Result: fmt.Sprintf(`%#x`, reply)}
- }
- return &jsonSuccessResponse{Version: jsonrpcVersion, Id: id, Result: reply}
- }
- // CreateErrorResponse will create a JSON-RPC error response with the given id and error.
- func (c *jsonCodec) CreateErrorResponse(id interface{}, err Error) interface{} {
- return &jsonErrResponse{Version: jsonrpcVersion, Id: id, Error: jsonError{Code: err.ErrorCode(), Message: err.Error()}}
- }
- // CreateErrorResponseWithInfo will create a JSON-RPC error response with the given id and error.
- // info is optional and contains additional information about the error. When an empty string is passed it is ignored.
- func (c *jsonCodec) CreateErrorResponseWithInfo(id interface{}, err Error, info interface{}) interface{} {
- return &jsonErrResponse{Version: jsonrpcVersion, Id: id,
- Error: jsonError{Code: err.ErrorCode(), Message: err.Error(), Data: info}}
- }
- // CreateNotification will create a JSON-RPC notification with the given subscription id and event as params.
- func (c *jsonCodec) CreateNotification(subid, namespace string, event interface{}) interface{} {
- if isHexNum(reflect.TypeOf(event)) {
- return &jsonNotification{Version: jsonrpcVersion, Method: namespace + notificationMethodSuffix,
- Params: jsonSubscription{Subscription: subid, Result: fmt.Sprintf(`%#x`, event)}}
- }
- return &jsonNotification{Version: jsonrpcVersion, Method: namespace + notificationMethodSuffix,
- Params: jsonSubscription{Subscription: subid, Result: event}}
- }
- // Write message to client
- func (c *jsonCodec) Write(res interface{}) error {
- c.encMu.Lock()
- defer c.encMu.Unlock()
- return c.encode(res)
- }
- // Close the underlying connection
- func (c *jsonCodec) Close() {
- c.closer.Do(func() {
- close(c.closed)
- c.rw.Close()
- })
- }
- // Closed returns a channel which will be closed when Close is called
- func (c *jsonCodec) Closed() <-chan interface{} {
- return c.closed
- }
|