rpc ethereum client protocol RPC integration

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

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

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

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])
+	}
+}

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

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