From 98d3b24637fb3166cc2905bd293a17ef65fc9bc4 Mon Sep 17 00:00:00 2001 From: Bas van Kervel Date: Tue, 29 Sep 2015 10:50:00 +0200 Subject: [PATCH] rpc ethereum client protocol RPC integration --- rpc/ecp/doc.go | 75 ++++++++ rpc/ecp/errors.go | 119 ++++++++++++ rpc/ecp/protocol.go | 380 +++++++++++++++++++++++++++++++++++++++ rpc/ecp/protocol_test.go | 107 +++++++++++ rpc/ecp/server.go | 258 ++++++++++++++++++++++++++ rpc/ecp/utils.go | 167 +++++++++++++++++ 6 files changed, 1106 insertions(+) create mode 100644 rpc/ecp/doc.go create mode 100644 rpc/ecp/errors.go create mode 100644 rpc/ecp/protocol.go create mode 100644 rpc/ecp/protocol_test.go create mode 100644 rpc/ecp/server.go create mode 100644 rpc/ecp/utils.go diff --git a/rpc/ecp/doc.go b/rpc/ecp/doc.go new file mode 100644 index 0000000000..b0fa33f2fe --- /dev/null +++ b/rpc/ecp/doc.go @@ -0,0 +1,75 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +/* +Package ecp implements the Ethereum Client Protocol (ECP) and contains a server to +create a RPC server offering a public API using ECP. + +The purpose of this package is to provide access to exported methods an I/O connection. +A client can instantiate a server and register objects. The server will expose the +object as a service. After registration exported methods can be called remotely. + +Methods that satisfy the following criteria will be exposed: + - the object is exported + - the method is exported + - arguments must be builtin types or *big.Int + - when the method returns an error its must be the last returned values + +The REdis Serialization Protocol (http://redis.io/topics/protocol) is used as +serialization format. It is a fast and easy to parse protocol which is binary safe. +Currently the following types are supported: + + - string, send as simple string + - []byte, send as bulk/binary string + - big.Int, send as bulk/binary string in big endian format + - int64, send as integer + - error, as the default error type starting with an error code following the error message + - struct, fields are send in an array in the definition order + - array, as an array + +Nil values are send as a null array in case the type is an array, otherwise as a null +binary string. + +See the official REdis Serialization Protocol specification page for a description and +examples of how data is serialized. The ecp server uses the same convention as the Redis +server for messages. Requests and responses are represented in an array. Services who return +no data return with the +OK\r\n message as an indication that the request was processed. + +Example server: + + type ExampleService struct { + } + + func (s *ExampleService) Echo(str string, i int) (string, int) { + return str, i + } + + func main() { + srvr := ecp.NewServer() + srvr.RegisterName("example", new(ExampleService)) + + l, _ := net.ListenUnix("unix", &net.UnixAddr{Name: "/tmp/example.sock", Net: "unix"}) + go srvr.Serve(l) + ... + srvr.Stop() + } + +A client can now send the following request: + *3\r\n+example.Echo\r\n+some string\r\n:42\r\n +And receive the following response: + *2\r\n+some string\r\n:42\r\n +*/ +package ecp diff --git a/rpc/ecp/errors.go b/rpc/ecp/errors.go new file mode 100644 index 0000000000..c33c78c193 --- /dev/null +++ b/rpc/ecp/errors.go @@ -0,0 +1,119 @@ +package ecp + +import ( + "fmt" + "reflect" +) + +type errorCode int + +const ( + _ errorCode = iota + ProtocolError // Received message invalid + InvalidRequestError // Received message isn't a valid request + MethodNotFoundError // Requested method not found + InvalidArguments // Received arguments are invalid + LogicError // Service returned an error +) + +// isRecoverable returns an indication if the server can recover from the error +func isRecoverable(err error) bool { + switch err.(type) { + case *unknownServiceError, *unknownMethodError, *invalidNumberOfArgumentsError, *invalidStructArgumentError, + *invalidArgumentError, *invalidRequestError, *unsupportedTypeError, *callbackError: + return true + } + + return false +} + +// InvalidLeadInError is raised when the received message doesn't for a valid +// RESP message. It is a bit more detailed than InvalidRequestError. +type invalidLeadInError struct { + got byte +} + +func (e *invalidLeadInError) Error() string { + return fmt.Sprintf("%d ECP error, received unexpected byte 0x%x", ProtocolError, e.got) +} + +// UnexpectedByteError occurs when incoming data doesn't follow the protocol specifications +type unexpectedByteError struct { + expected, got byte +} + +func (e *unexpectedByteError) Error() string { + return fmt.Sprintf("%d ECP error, expected 0x%x got 0x%x", ProtocolError, e.expected, e.got) +} + +type unknownServiceError struct { + svc string +} + +func (e *unknownServiceError) Error() string { + return fmt.Sprintf("%d service '%s' not found", MethodNotFoundError, e.svc) +} + +type unknownMethodError struct { + svc string + method string +} + +func (e *unknownMethodError) Error() string { + return fmt.Sprintf("%d method '%s.%s' not found", MethodNotFoundError, e.svc, e.method) +} + +type invalidNumberOfArgumentsError struct { + svc string + method string + exp int + got int +} + +func (e *invalidNumberOfArgumentsError) Error() string { + return fmt.Sprintf("%d %s.%s expect %d argument(s) but got %d", + InvalidArguments, e.svc, e.method, e.exp, e.got) +} + +type invalidArgumentError struct { + pos int + exp reflect.Type + got reflect.Type +} + +func (e *invalidArgumentError) Error() string { + return fmt.Sprintf("%d expected type for argument %d is %s but got %s", InvalidArguments, e.pos, e.exp, e.got) +} + +type invalidStructArgumentError struct { + to reflect.Type + exp int + got int +} + +func (e *invalidStructArgumentError) Error() string { + return fmt.Sprintf("%s expected %d values but got %d", e.to.Name(), e.exp, e.got) +} + +type invalidRequestError struct { +} + +func (e *invalidRequestError) Error() string { + return fmt.Sprintf("%d invalid request", InvalidRequestError) +} + +type unsupportedTypeError struct { + name string +} + +func (e *unsupportedTypeError) Error() string { + return fmt.Sprintf("%d '%s' is an unsupported type", ProtocolError, e.name) +} + +type callbackError struct { + msg string +} + +func (e *callbackError) Error() string { + return fmt.Sprintf("%d %s", LogicError, e.msg) +} diff --git a/rpc/ecp/protocol.go b/rpc/ecp/protocol.go new file mode 100644 index 0000000000..b0b187792b --- /dev/null +++ b/rpc/ecp/protocol.go @@ -0,0 +1,380 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package ecp + +import ( + "bufio" + "fmt" + "io" + "math/big" + "reflect" + "strconv" + "strings" +) + +const ( + firstDelim = '\r' + lastDelim = '\n' + delim = "\r\n" + simpleStrMark = '+' + errMark = '-' + intMark = ':' + bulkStrMark = '$' + arrayMark = '*' + + methodSep = "." + successReply = "+OK\r\n" +) + +type ValueType int + +const ( + simpleStrValue ValueType = 1 << iota + integerValue + binaryValue + arrayValue + errorValue + nullValue +) + +// ServerCodec implements reading and writing RPC messages for the server side of a RPC session. +type ServerCodec interface { + // Read the next request + Read() (*request, error) + // WriteResponse writes an array of values as a response to the client + WriteResponse([]interface{}) error + // WriteError writes an error to the client + WriteError(error) error + // Close the underlying data stream(s) + Close() error +} + +// ECPCodec implements the REdis Server Protocol (RESP). +// See for more information http://redis.io/topics/protocol. +type ECPCodec struct { + rc io.ReadCloser + r *bufio.Reader + wc io.WriteCloser + w *bufio.Writer +} + +// Create a new Ethereum Client Protocol Codec. It reads incoming requests from the supplied +// reader and writes responses to the supplied writer. +func NewECPCodec(r io.ReadCloser, w io.WriteCloser) ServerCodec { + return &ECPCodec{r, bufio.NewReader(r), w, bufio.NewWriter(w)} +} + +type message struct { + Kind ValueType + Val interface{} +} + +// Close underlying reader and writer +func (c *ECPCodec) Close() error { + if err := c.w.Flush(); err != nil { + return err + } + + c.rc.Close() + c.wc.Close() + + return nil +} + +// Read reads a new RPC request. +func (c *ECPCodec) Read() (*request, error) { + msg, err := c.read() + if err != nil { + return nil, err + } + + if msg.Kind == arrayValue { + if values, ok := msg.Val.([]*message); ok { + if len(values) > 0 { + req := new(request) + method := strings.Split(fmt.Sprintf("%s", values[0].Val), methodSep) + if len(method) != 2 { + return nil, &invalidRequestError{} + } + + req.service, req.method = method[0], method[1] + req.args = make([]interface{}, len(values)-1) + for i := 1; i < len(values); i++ { + req.args[i-1] = values[i].Val + } + + return req, nil + } + } + } + + return nil, &invalidRequestError{} +} + +func (c *ECPCodec) read() (*message, error) { + mark, err := c.r.ReadByte() + if err != nil { + return nil, err + } + + switch mark { + case simpleStrMark: + return c.readSimpleStr() + case bulkStrMark: + return c.readBulkStr() + case intMark: + return c.readInteger() + case arrayMark: + return c.readArray() + case errMark: + return c.readError() + } + + return nil, &invalidLeadInError{mark} +} + +func (c *ECPCodec) readSimpleStr() (*message, error) { + str, err := c.r.ReadBytes(firstDelim) + if err != nil { + return nil, err + } + + if nl, err := c.r.ReadByte(); err != nil { + return nil, err + } else if nl != lastDelim { + return nil, &unexpectedByteError{lastDelim, nl} + } + + return &message{simpleStrValue, string(str[:len(str)-1])}, nil +} + +func (c *ECPCodec) readError() (*message, error) { + e, err := c.r.ReadBytes(firstDelim) + if err != nil { + return nil, err + } + + if nl, err := c.r.ReadByte(); err != nil { + return nil, err + } else if nl != lastDelim { + return nil, &unexpectedByteError{lastDelim, nl} + } + + return &message{errorValue, string(e[:len(e)-1])}, nil +} + +func (c *ECPCodec) readInteger() (*message, error) { + i, err := c.r.ReadBytes(firstDelim) + if err != nil { + return nil, err + } + + if nl, err := c.r.ReadByte(); err != nil { + return nil, err + } else if nl != lastDelim { + return nil, &unexpectedByteError{lastDelim, nl} + } + + val, err := strconv.ParseInt(string(i[:len(i)-1]), 10, 64) + if err != nil { + return nil, err + } + return &message{integerValue, val}, nil +} + +func (c *ECPCodec) readBulkStr() (*message, error) { + n, err := c.r.ReadBytes(firstDelim) + if err != nil { + return nil, err + } + + nl, err := c.r.ReadByte() + if err != nil { + return nil, err + } else if nl != lastDelim { + return nil, &unexpectedByteError{lastDelim, nl} + } + + cnt, err := strconv.ParseInt(string(n[:len(n)-1]), 10, 64) + if err != nil { + return nil, err + } + + if cnt == -1 { + return &message{nullValue, nil}, nil + } + if cnt < 0 { + return nil, &invalidRequestError{} + } + + val := make([]byte, cnt) + tot := int64(0) + for tot < cnt { + n, err := c.r.Read(val[tot:]) + if err != nil { + return nil, err + } + tot += int64(n) + } + + if cr, err := c.r.ReadByte(); err != nil { + return nil, err + } else if cr != firstDelim { + return nil, &unexpectedByteError{lastDelim, cr} + } + + if nl, err := c.r.ReadByte(); err != nil { + return nil, err + } else if nl != lastDelim { + return nil, &unexpectedByteError{lastDelim, nl} + } + + return &message{binaryValue, val}, nil +} + +func (c *ECPCodec) readArray() (*message, error) { + n, err := c.r.ReadBytes(firstDelim) + if err != nil { + return nil, err + } + + if nl, err := c.r.ReadByte(); err != nil { + return nil, err + } else if nl != lastDelim { + return nil, &unexpectedByteError{lastDelim, nl} + } + + nElem, err := strconv.ParseInt(string(n[:len(n)-1]), 10, 64) + if err != nil { + return nil, err + } + + arr := make([]*message, nElem) + + for i := int64(0); i < nElem; i++ { + if arr[i], err = c.read(); err != nil { + return nil, err + } + } + + return &message{arrayValue, arr}, nil +} + +// WriteResponse writes a values to the client. When values is nil the default +// success response is written. +func (c *ECPCodec) WriteResponse(values []interface{}) error { + defer c.w.Flush() + + if values == nil { + _, e := c.w.WriteString(successReply) + return e + } + + return c.writeArray(values) +} + +// WriteError writes the supplied error to the client +func (c *ECPCodec) WriteError(err error) error { + defer c.w.Flush() + _, e := c.w.WriteString(fmt.Sprintf("%c%s%s", errMark, err, delim)) + return e +} + +func (c *ECPCodec) writeSimpleString(str string) error { + _, err := c.w.WriteString(fmt.Sprintf("%c%s%s", simpleStrMark, str, delim)) + return err +} + +func (c *ECPCodec) writeInteger(i int64) error { + _, err := c.w.WriteString(fmt.Sprintf("%c%d%s", intMark, i, delim)) + return err +} + +func (c *ECPCodec) writeBinary(bin []byte) error { + if bin == nil { + _, err := c.w.WriteString(fmt.Sprintf("%c-1%s", bulkStrMark, delim)) + return err + } + + if _, err := c.w.WriteString(fmt.Sprintf("%c%d%s", bulkStrMark, len(bin), delim)); err != nil { + return err + } + + if _, err := c.w.Write(bin); err != nil { + return err + } + + _, err := c.w.WriteString(delim) + return err +} + +func (c *ECPCodec) writeStruct(obj interface{}) error { + val := reflect.ValueOf(obj) + fields := make([]interface{}, val.NumField()) + cnt := 0 + for i := 0; i < val.NumField(); i++ { + if val.Field(i).CanInterface() { // skip unexported fields + fields[cnt] = val.Field(i).Interface() + cnt++ + } + } + return c.writeArray(fields[:cnt]) +} + +func (c *ECPCodec) writeArray(arr []interface{}) (err error) { + if arr == nil { + _, err = c.w.WriteString(fmt.Sprintf("%c-1%s", arrayMark, delim)) + return + } + + // array length prefix + if _, err = c.w.WriteString(fmt.Sprintf("%c%d%s", arrayMark, len(arr), delim)); err != nil { + return + } + + for i := 0; i < len(arr) && err == nil; i++ { + switch val := arr[i].(type) { + case string: + err = c.writeSimpleString(val) + case *string: + err = c.writeSimpleString(*val) + case []byte: + err = c.writeBinary(val) + case big.Int: + err = c.writeBinary(val.Bytes()) + case *big.Int: + err = c.writeBinary(val.Bytes()) + case int64: + err = c.writeInteger(val) + case nil: // use nil bulk string to represent nil values + err = c.writeBinary(nil) + case error: + err = c.WriteError(val) + default: + typ := reflect.TypeOf(val) + if typ.Kind() == reflect.Struct { + err = c.writeStruct(val) + } else if a, ok := val.([]interface{}); ok { + err = c.writeArray(a) + } else { + err = &unsupportedTypeError{typ.Name()} + } + } + } + + return +} diff --git a/rpc/ecp/protocol_test.go b/rpc/ecp/protocol_test.go new file mode 100644 index 0000000000..a8daaba24f --- /dev/null +++ b/rpc/ecp/protocol_test.go @@ -0,0 +1,107 @@ +package ecp + +import ( + "bytes" + "fmt" + "io" + "math/big" + "strings" + "testing" +) + +type testReader struct { + r io.Reader +} + +func (r *testReader) Read(p []byte) (n int, err error) { + return r.r.Read(p) +} + +func (r *testReader) Close() error { + return nil +} + +func newTestReader(source string) io.ReadCloser { + return &testReader{strings.NewReader(source)} +} + +func TestNoArgsAndRets(t *testing.T) { + r := newTestReader("*1\r\n+Service.Method\r\n") + codec := NewECPCodec(r, nil) + + req, err := codec.Read() + if err != nil { + t.Fatalf("%s\n", err) + } + + if req.service != "Service" { + t.Fatalf("Expected service 'Service' but got '%s'\n", req.service) + } + + if req.method != "Method" { + t.Fatalf("Expected method 'Method' but got '%s'\n", req.method) + } +} + +func TestWithArgs(t *testing.T) { + i := big.NewInt(123456789) + ib := i.Bytes() + str := "some string" + bstr := []byte{'a', '\r', '\n', 'b'} + in := int64(42) + + r := newTestReader(fmt.Sprintf("*5\r\n+Service.Method\r\n$%d\r\n%s\r\n+%s\r\n$%d\r\n%s\r\n:%d\r\n", + len(ib), ib, str, len(bstr), bstr, in)) + + codec := NewECPCodec(r, nil) + + req, err := codec.Read() + if err != nil { + t.Fatalf("%s\n", err) + } + + if req.service != "Service" { + t.Fatalf("Expected service 'Service' but got '%s'\n", req.service) + } + + if req.method != "Method" { + t.Fatalf("Expected method 'Method' but got '%s'\n", req.method) + } + + if len(req.args) != 4 { + t.Fatalf("Expected 3 args, got %d args(s)\n", len(req.args)) + } + + if pin, ok := req.args[0].([]byte); ok { + parsedInt := new(big.Int).SetBytes(pin) + if parsedInt.Cmp(i) != 0 { + t.Fatalf("Expected big.Int(%d).Cmp(%d) == 0\n", parsedInt, i) + } + } else { + t.Fatalf("Expected arg[0] to be []byte, got %T\n", req.args[0]) + } + + if s, ok := req.args[1].(string); ok { + if str != s { + t.Fatalf("Expected 'some string' == '%s'\n", s) + } + } else { + t.Fatalf("Expected arg[1] to be string, got %T\n", req.args[1]) + } + + if arr, ok := req.args[2].([]byte); ok { + if bytes.Compare(arr, bstr) != 0 { + t.Fatalf("Expected %s, got %s\n", bstr, arr) + } + } else { + t.Fatalf("Expected arg[2] to be []byte, got %T\n", req.args[2]) + } + + if val, ok := req.args[3].(int64); ok { + if in != val { + t.Fatalf("Expected %d, got %d\n", in, val) + } + } else { + t.Fatalf("Expected arg[3] to be int64, got %T\n", req.args[3]) + } +} diff --git a/rpc/ecp/server.go b/rpc/ecp/server.go new file mode 100644 index 0000000000..beac8602e1 --- /dev/null +++ b/rpc/ecp/server.go @@ -0,0 +1,258 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package ecp + +import ( + "fmt" + "net" + "reflect" + "sync" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" +) + +type request struct { + service string // registered service name + method string // callback on service + args []interface{} // arguments for callback +} + +type callback struct { + method reflect.Method // callback + args []reflect.Type // input argument types + errPos int // err return idx, of -1 when method cannot return error +} + +type service struct { + name string // name for service + rcvr reflect.Value // receiver of methods for the service + typ reflect.Type // receiver type + callbacks callbacks // registered handlers +} + +type serviceRegistry map[string]*service +type codecCollection map[ServerCodec]interface{} +type callbacks map[string]*callback + +// Server represents a RPC server +type Server struct { + listener net.Listener + serviceMap serviceRegistry + mu sync.Mutex // protects codec collection + codecs codecCollection // active connections/codecs +} + +// NewServer will create a new server instance with no registered handlers. +func NewServer() *Server { + return &Server{ + serviceMap: make(serviceRegistry), + codecs: make(codecCollection), + } +} + +// Register publishes suitable methods of rcvr. Methods will be published +// with a dot between the rcrv and method, e.g. ChainManager.GetBlock. +func (s *Server) Register(rcvr interface{}) error { + return s.register(rcvr, "", false) +} + +// Register publishes suitable methods of rcvr. Methods will be published +// with a dot between the given name and method, e.g. ChainManager.GetBlock. +func (s *Server) RegisterName(name string, rcvr interface{}) error { + return s.register(rcvr, name, true) +} + +func (s *Server) register(rcvr interface{}, name string, useName bool) error { + if s.serviceMap == nil { + s.serviceMap = make(map[string]*service) + } + + svc := new(service) + svc.typ = reflect.TypeOf(rcvr) + svc.rcvr = reflect.ValueOf(rcvr) + + sname := reflect.Indirect(svc.rcvr).Type().Name() + if useName { + sname = name + } + if sname == "" { + return fmt.Errorf("no service name for type %s", svc.typ.String()) + } + if !isExported(sname) && !useName { + return fmt.Errorf("%s is not exported", sname) + } + + if _, present := s.serviceMap[sname]; present { + return fmt.Errorf("%s already registered", sname) + } + + svc.name = sname + svc.callbacks = suitableCallbacks(svc.typ) + + if len(svc.callbacks) == 0 { + return fmt.Errorf("Service doesn't have any suitable methods to expose") + } + + s.serviceMap[svc.name] = svc + + return nil +} + +// Serve accepts connections and starts processing incoming requests. +func (s *Server) Serve(l net.Listener) { + s.listener = l + for { + c, err := s.listener.Accept() + if err != nil { + glog.V(logger.Debug).Infof("%v\n", err) + break + } + + glog.V(logger.Debug).Infof("Accepted connection from %s\n", c.RemoteAddr()) + codec := NewECPCodec(c, c) + s.mu.Lock() + s.codecs[codec] = nil + s.mu.Unlock() + go s.serveCodec(codec) + } + + // stopped accepting new connections, close existing + for c, _ := range s.codecs { + s.close(c) + } +} + +// Stop will stop listening for new connection and closes existing connections +func (s *Server) Stop() error { + if s.listener != nil { + return s.listener.Close() + } + return nil +} + +// serveCodec start the server and uses the supplied codec to read and write +// requests and response. It will call Close on the codec when it returns. +func (s *Server) serveCodec(codec ServerCodec) error { + defer s.close(codec) + + for { + svc, c, argv, err := s.readRequest(codec) + if err != nil { + err2 := codec.WriteError(err) + if !isRecoverable(err) { + glog.V(logger.Error).Infoln(err) + return err + } + if !isRecoverable(err2) { + glog.V(logger.Error).Infoln(err2) + return err2 + } + continue + } + + if res, err := s.call(c, svc.rcvr, argv); err != nil { + if err = codec.WriteError(err); err != nil { + if !isRecoverable(err) { + glog.V(logger.Error).Infoln(err) + return err + } + } + } else { + if res == nil { // callback has no return values + if err = codec.WriteResponse(nil); err != nil { + if !isRecoverable(err) { + glog.V(logger.Error).Infoln(err) + return err + } + } + } else { + values := make([]interface{}, len(res)) + for i := 0; i < len(res); i++ { + values[i] = res[i].Interface() + } + + if err = codec.WriteResponse(values); err != nil { + if !isRecoverable(err) { + glog.V(logger.Error).Infoln(err) + return err + } + } + } + } + } +} + +func (s *Server) close(codec ServerCodec) { + codec.Close() + s.mu.Lock() + delete(s.codecs, codec) + s.mu.Unlock() +} + +func (s *Server) call(c *callback, rcvr reflect.Value, argv []reflect.Value) ([]reflect.Value, error) { + args := make([]reflect.Value, 1+len(argv)) + args[0] = rcvr + for i := 0; i < len(argv); i++ { + args[i+1] = argv[i] + } + rets := c.method.Func.Call(args) + + // check for error + if c.errPos >= 0 && !rets[c.errPos].IsNil() { + return nil, &callbackError{fmt.Sprintf("%s", rets[c.errPos])} + } + + // don't send error when there is none + if c.errPos > 0 { + return rets[:c.errPos], nil + } + + return rets, nil +} + +func (s *Server) readRequest(codec ServerCodec) (svc *service, c *callback, argv []reflect.Value, err error) { + req, err := codec.Read() + if err != nil { + return nil, nil, nil, err + } + + // determine callback + if svc = s.serviceMap[req.service]; svc == nil { + return nil, nil, nil, &unknownServiceError{req.service} + } + + if c = svc.callbacks[req.method]; c == nil { + return nil, nil, nil, &unknownMethodError{req.service, req.method} + } + + // decode params + nArgs := len(c.args) + if nArgs != len(req.args) { + return nil, nil, nil, &invalidNumberOfArgumentsError{req.service, req.method, nArgs, len(req.args)} + } + + argv = make([]reflect.Value, nArgs) + for i := 0; i < nArgs; i++ { + argv[i], err = convert(i, req.args[i], c.args[i]) + if err != nil { + return nil, nil, nil, err + } + } + + return +} diff --git a/rpc/ecp/utils.go b/rpc/ecp/utils.go new file mode 100644 index 0000000000..d140abf5dc --- /dev/null +++ b/rpc/ecp/utils.go @@ -0,0 +1,167 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package ecp + +import ( + "math/big" + "reflect" + "strconv" + "unicode" + "unicode/utf8" +) + +// Is this an exported - upper case - name? +func isExported(name string) bool { + rune, _ := utf8.DecodeRuneInString(name) + return unicode.IsUpper(rune) +} + +// Is this type exported or a builtin? +func isExportedOrBuiltinType(t reflect.Type) bool { + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + // PkgPath will be non-empty even for an exported type, + // so we need to check the type name as well. + return isExported(t.Name()) || t.PkgPath() == "" +} + +var errorType = reflect.TypeOf((*error)(nil)).Elem() + +// Implements this type the error interface +func isErrorType(t reflect.Type) bool { + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + return t.Implements(errorType) +} + +func suitableCallbacks(typ reflect.Type) callbacks { + callbacks := make(callbacks) +METHODS: + for m := 0; m < typ.NumMethod(); m++ { + method := typ.Method(m) + mtype := method.Type + mname := method.Name + if method.PkgPath != "" { // method must be exported + continue + } + + var h callback + h.method = method + numIn := mtype.NumIn() + + // determine arguments, ignore first arg since it's the receiver type + h.args = make([]reflect.Type, numIn-1) + for i := 1; i < numIn; i++ { + argType := mtype.In(i) + if !isExportedOrBuiltinType(argType) { + continue METHODS + } + h.args[i-1] = argType + } + + // determine if callback can return an error + h.errPos = -1 + for i := 0; i < mtype.NumOut(); i++ { + if isErrorType(mtype.Out(i)) { + h.errPos = i + break + } + } + + // only support methods which return an error as the last returned value (or no error). + if h.errPos == -1 || (h.errPos >= 0 && h.errPos == mtype.NumOut()-1) { + callbacks[mname] = &h + } + } + + return callbacks +} + +var bigIntType = reflect.TypeOf(new(big.Int)) + +func isBigInt(typ reflect.Type) bool { + return typ == bigIntType +} + +// in the future we might consider adding support for more conversion as the API gets defined +func convert(pos int, val interface{}, to reflect.Type) (reflect.Value, error) { + from := reflect.TypeOf(val) + if from.Kind() == to.Kind() { + return reflect.ValueOf(val), nil + } + + plainType := to + for plainType.Kind() == reflect.Ptr { + plainType = plainType.Elem() + } + + if plainType.Kind() == reflect.Struct { + return convertStruct(pos, val, to, plainType) + } + + retPtr := reflect.New(plainType) + if from.ConvertibleTo(plainType) { + ret := reflect.Indirect(retPtr) + ret.Set(reflect.ValueOf(val).Convert(plainType)) + if to.Kind() == reflect.Ptr { + return retPtr, nil + } + return ret, nil + } + + if bytes, ok := val.([]byte); ok && to.Kind() == reflect.Int64 { + if i, err := strconv.ParseInt(string(bytes), 10, 64); err == nil { + return reflect.ValueOf(i), nil + } + } + + if bytes, ok := val.([]byte); ok && isBigInt(to) { + v := new(big.Int) + v.SetBytes(bytes) + return reflect.ValueOf(v), nil + } + + return reflect.Zero(to), &invalidArgumentError{1 + pos, to, reflect.TypeOf(val)} +} + +func convertStruct(pos int, val interface{}, to, plainType reflect.Type) (reflect.Value, error) { + if fields, ok := val.([]*message); ok { + if len(fields) != plainType.NumField() { + return reflect.Zero(to), &invalidStructArgumentError{to, plainType.NumField(), len(fields)} + } + + ptrVal := reflect.New(plainType) + v := reflect.Indirect(ptrVal) + for i := 0; i < v.NumField(); i++ { + val, err := convert(pos, fields[i].Val, v.Field(i).Type()) + if err != nil { + return reflect.Zero(to), err + } + v.Field(i).Set(val) + } + + if to.Kind() == reflect.Ptr { + return ptrVal, nil + } + + return v, nil + } + + return reflect.Zero(to), &invalidArgumentError{1 + pos, to, reflect.TypeOf(val)} +}