rpc/v2: new RPC implementation with pub/sub support

common/types.go

@@ -21,6 +21,7 @@ import (
 	"math/big"
 	"math/rand"
 	"reflect"
+	"encoding/json"
 )
 
 const (
@@ -50,6 +51,11 @@ func (h Hash) Bytes() []byte { return h[:] }
 func (h Hash) Big() *big.Int { return Bytes2Big(h[:]) }
 func (h Hash) Hex() string   { return "0x" + Bytes2Hex(h[:]) }
 
+// Serialize given hash to JSON
+func (h Hash) MarshalJSON() ([]byte, error) {
+	return json.Marshal(h.Hex())
+}
+
 // Sets the hash to the value of b. If b is larger than len(h) it will panic
 func (h *Hash) SetBytes(b []byte) {
 	if len(b) > len(h) {
@@ -117,6 +123,11 @@ func (a *Address) Set(other Address) {
 	}
 }
 
+// Serialize given address to JSON
+func (a Address) MarshalJSON() ([]byte, error) {
+	return json.Marshal(a.Hex())
+}
+
 // PP Pretty Prints a byte slice in the following format:
 // 	hex(value[:4])...(hex[len(value)-4:])
 func PP(value []byte) string {

ethdb/interface.go

@@ -28,3 +28,4 @@ type Batch interface {
 	Put(key, value []byte) error
 	Write() error
 }
+

rpc/v2/doc.go

@@ -0,0 +1,102 @@
+// 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 provides access to the exported methods of an object across a network
+or other I/O connection. After creating a server instance objects can be registered,
+making it visible from the outside. Exported methods that follow specific
+conventions can be called remotely. It also has support for the publish/subscribe
+pattern.
+
+Methods that satisfy the following criteria are made available for remote access:
+ - object must be exported
+ - method must be exported
+ - method returns 0, 1 (response or error) or 2 (response and error) values
+ - method argument(s) must be exported or builtin types
+ - method returned value(s) must be exported or builtin types
+
+An example method:
+ func (s *CalcService) Div(a, b int) (int, error)
+
+When the returned error isn't nil the returned integer is ignored and the error is
+send back to the client. Otherwise the returned integer is send back to the client.
+
+The server offers the ServeCodec method which accepts a ServerCodec instance. It will
+read requests from the codec, process the request and sends the response back to the
+client using the codec. The server can execute requests concurrently. Responses
+can be send back to the client out of order.
+
+An example server which uses the JSON codec:
+ type CalculatorService struct {}
+
+ func (s *CalculatorService) Add(a, b int) int {
+	return a + b
+ }
+
+ func (s *CalculatorService Div(a, b int) (int, error) {
+	if b == 0 {
+		return 0, errors.New("divide by zero")
+	}
+	return a/b, nil
+ }
+
+ calculator := new(CalculatorService)
+ server := NewServer()
+ server.RegisterName("calculator", calculator")
+
+ l, _ := net.ListenUnix("unix", &net.UnixAddr{Net: "unix", Name: "/tmp/calculator.sock"})
+ for {
+	c, _ := l.AcceptUnix()
+	codec := v2.NewJSONCodec(c)
+	go server.ServeCodec(codec)
+ }
+
+The package also supports the publish subscribe pattern through the use of subscriptions.
+A method that is considered eligible for notifications must satisfy the following criteria:
+ - object must be exported
+ - method must be exported
+ - method argument(s) must be exported or builtin types
+ - method must return the tuple Subscription, error
+
+
+An example method:
+ func (s *BlockChainService) Head() (Subscription, error) {
+ 	sub := s.bc.eventMux.Subscribe(ChainHeadEvent{})
+	return v2.NewSubscription(sub), nil
+ }
+
+This method will push all raised ChainHeadEvents to subscribed clients. If the client is only
+interested in every N'th block it is possible to add a criteria.
+
+ func (s *BlockChainService) HeadFiltered(nth uint64) (Subscription, error) {
+ 	sub := s.bc.eventMux.Subscribe(ChainHeadEvent{})
+
+	criteria := func(event interface{}) bool {
+		chainHeadEvent := event.(ChainHeadEvent)
+		if chainHeadEvent.Block.NumberU64() % nth == 0 {
+			return true
+		}
+		return false
+	}
+
+	return v2.NewSubscriptionFiltered(sub, criteria), nil
+ }
+
+Subscriptions are deleted when:
+ - the user sends an unsubscribe request
+ - the connection which was used to create the subscription is closed
+*/
+package v2

rpc/v2/errors.go

@@ -0,0 +1,85 @@
+// 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 v2
+
+import "fmt"
+
+// request is for an unknown service
+type methodNotFoundError struct {
+	service string
+	method  string
+}
+
+func (e *methodNotFoundError) Code() int {
+	return -32601
+}
+
+func (e *methodNotFoundError) Error() string {
+	return fmt.Sprintf("The method %s%s%s does not exist/is not available", e.service, serviceMethodSeparator, e.method)
+}
+
+// received message isn't a valid request
+type invalidRequestError struct {
+	message string
+}
+
+func (e *invalidRequestError) Code() int {
+	return -32600
+}
+
+func (e *invalidRequestError) Error() string {
+	return e.message
+}
+
+// received message is invalid
+type invalidMessageError struct {
+	message string
+}
+
+func (e *invalidMessageError) Code() int {
+	return -32700
+}
+
+func (e *invalidMessageError) Error() string {
+	return e.message
+}
+
+// unable to decode supplied params, or an invalid number of parameters
+type invalidParamsError struct {
+	message string
+}
+
+func (e *invalidParamsError) Code() int {
+	return -32602
+}
+
+func (e *invalidParamsError) Error() string {
+	return e.message
+}
+
+// logic error, callback returned an error
+type callbackError struct {
+	message string
+}
+
+func (e *callbackError) Code() int {
+	return -32000
+}
+
+func (e *callbackError) Error() string {
+	return e.message
+}

rpc/v2/json.go

@@ -0,0 +1,295 @@
+// 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 v2
+
+import (
+	"encoding/json"
+	"io"
+	"reflect"
+	"strings"
+	"sync/atomic"
+)
+
+const (
+	jsonRPCVersion         = "2.0"
+	serviceMethodSeparator = "_"
+	subscribeMethod        = "eth_subscribe"
+	unsubscribeMethod      = "eth_unsubscribe"
+	notificationMethod     = "eth_subscription"
+)
+
+// JSON-RPC request
+type jsonRequest struct {
+	Method  string          `json:"method"`
+	Version string          `json:"jsonrpc"`
+	Id      int64           `json:"id"`
+	Payload json.RawMessage `json:"params"`
+}
+
+// JSON-RPC response
+type jsonSuccessResponse struct {
+	Version string      `json:"jsonrpc"`
+	Id      int64       `json:"id"`
+	Result  interface{} `json:"result,omitempty"`
+}
+
+// JSON-RPC error object
+type jsonError struct {
+	Code    int         `json:"code"`
+	Message string      `json:"message"`
+	Data    interface{} `json:"data,omitempty"`
+}
+
+// JSON-RPC error response
+type jsonErrResponse struct {
+	Version string    `json:"jsonrpc"`
+	Id      *int64    `json:"id,omitempty"`
+	Error   jsonError `json:"error"`
+}
+
+// JSON-RPC notification payload
+type jsonSubscription struct {
+	Subscription string      `json:"subscription"`
+	Result       interface{} `json:"result,omitempty"`
+}
+
+// JSON-RPC notification
+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 {
+	closed   chan interface{}
+	isClosed int32
+	d        *json.Decoder
+	e        *json.Encoder
+	req      jsonRequest
+	rw       io.ReadWriteCloser
+}
+
+// NewJSONCodec creates a new RPC server codec with support for JSON-RPC 2.0
+func NewJSONCodec(rwc io.ReadWriteCloser) ServerCodec {
+	d := json.NewDecoder(rwc)
+	d.UseNumber()
+	return &jsonCodec{closed: make(chan interface{}), d: d, e: json.NewEncoder(rwc), rw: rwc, isClosed: 0}
+}
+
+// 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, RPCError) {
+	var incomingMsg json.RawMessage
+	if err := c.d.Decode(&incomingMsg); err != nil {
+		return nil, false, &invalidRequestError{err.Error()}
+	}
+
+	if isBatch(incomingMsg) {
+		return parseBatchRequest(incomingMsg)
+	}
+
+	return parseRequest(incomingMsg)
+}
+
+// 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, RPCError) {
+	var in jsonRequest
+	if err := json.Unmarshal(incomingMsg, &in); err != nil {
+		return nil, false, &invalidMessageError{err.Error()}
+	}
+
+	// subscribe are special, they will always use `subscribeMethod` as service method
+	if in.Method == subscribeMethod {
+		reqs := []rpcRequest{rpcRequest{id: in.Id, isPubSub: true}}
+		if len(in.Payload) > 0 {
+			// first param must be service.method name
+			subscriptionMethod := []reflect.Type{reflect.TypeOf("")}
+			if args, err := parsePositionalArguments(in.Payload, subscriptionMethod); err == nil && len(args) == 1 {
+				elems := strings.Split(args[0].String(), serviceMethodSeparator)
+				if len(elems) == 2 {
+					reqs[0].service, reqs[0].method = elems[0], elems[1]
+					reqs[0].params = in.Payload
+					return reqs, false, nil
+				}
+			}
+		}
+		return nil, false, &invalidRequestError{"Unable to parse subscription request"}
+	}
+
+	if in.Method == unsubscribeMethod {
+		return []rpcRequest{rpcRequest{id: in.Id, isPubSub: true,
+			method: unsubscribeMethod, params: in.Payload}}, false, nil
+	}
+
+	// regular RPC call
+	elems := strings.Split(in.Method, serviceMethodSeparator)
+	if len(elems) != 2 {
+		return nil, false, &methodNotFoundError{in.Method, ""}
+	}
+
+	if len(in.Payload) == 0 {
+		return []rpcRequest{rpcRequest{service: elems[0], method: elems[1], id: in.Id}}, false, nil
+	}
+
+	return []rpcRequest{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, RPCError) {
+	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 {
+		// (un)subscribe are special, they will always use the same service.method
+		if r.Method == subscribeMethod {
+			requests[i] = rpcRequest{id: r.Id, isPubSub: true}
+			if len(r.Payload) > 0 {
+				// first param must be service.method name
+				subscriptionMethod := []reflect.Type{reflect.TypeOf("")}
+				if args, err := parsePositionalArguments(r.Payload, subscriptionMethod); err == nil && len(args) == 1 {
+					elems := strings.Split(args[0].String(), serviceMethodSeparator)
+					if len(elems) == 2 {
+						requests[i].service, requests[i].method = elems[0], elems[1]
+						requests[i].params = r.Payload
+						continue
+					}
+				}
+			}
+
+			return nil, true, &invalidRequestError{"Unable to parse (un)subscribe request arguments"}
+		}
+
+		if r.Method == unsubscribeMethod {
+			requests[i] = rpcRequest{id: r.Id, isPubSub: true, method: unsubscribeMethod, params: r.Payload}
+			continue
+		}
+
+		elems := strings.Split(r.Method, serviceMethodSeparator)
+		if len(elems) != 2 {
+			return nil, true, &methodNotFoundError{r.Method, ""}
+		}
+
+		if len(r.Payload) == 0 {
+			requests[i] = rpcRequest{service: elems[0], method: elems[1], id: r.Id, params: nil}
+		} else {
+			requests[i] = rpcRequest{service: elems[0], method: elems[1], id: r.Id, params: r.Payload}
+		}
+	}
+
+	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, RPCError) {
+	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.
+func parsePositionalArguments(args json.RawMessage, argTypes []reflect.Type) ([]reflect.Value, RPCError) {
+	argValues := make([]reflect.Value, len(argTypes))
+	params := make([]interface{}, len(argTypes))
+	for i, t := range argTypes {
+		if t.Kind() == reflect.Ptr {
+			// values must be pointers for the Unmarshal method, reflect.
+			// Dereference otherwise reflect.New would create **SomeType
+			argValues[i] = reflect.New(t.Elem())
+			params[i] = argValues[i].Interface()
+		} else {
+			argValues[i] = reflect.New(t)
+			params[i] = argValues[i].Interface()
+		}
+	}
+
+	if err := json.Unmarshal(args, &params); err != nil {
+		return nil, &invalidParamsError{err.Error()}
+	}
+
+	// Convert pointers back to values where necessary
+	for i, a := range argValues {
+		if a.Kind() != argTypes[i].Kind() {
+			argValues[i] = reflect.Indirect(argValues[i])
+		}
+	}
+
+	return argValues, nil
+}
+
+// CreateResponse will create a JSON-RPC success response with the given id and reply as result.
+func (c *jsonCodec) CreateResponse(id int64, reply interface{}) interface{} {
+	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 *int64, err RPCError) interface{} {
+	return &jsonErrResponse{Version: jsonRPCVersion, Id: id, Error: jsonError{Code: err.Code(), 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 *int64, err RPCError, info interface{}) interface{} {
+	return &jsonErrResponse{Version: jsonRPCVersion, Id: id,
+		Error: jsonError{Code: err.Code(), 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 string, event interface{}) interface{} {
+	return &jsonNotification{Version: jsonRPCVersion, Method: notificationMethod,
+		Params: jsonSubscription{Subscription: subid, Result: event}}
+}
+
+// Write message to client
+func (c *jsonCodec) Write(res interface{}) error {
+	return c.e.Encode(res)
+}
+
+// Close the underlying connection
+func (c *jsonCodec) Close() {
+	if atomic.CompareAndSwapInt32(&c.isClosed, 0, 1) {
+		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
+}

rpc/v2/json_test.go

@@ -0,0 +1,73 @@
+package v2
+
+import (
+	"bufio"
+	"bytes"
+	"reflect"
+	"testing"
+)
+
+type RWC struct {
+	*bufio.ReadWriter
+}
+
+func (rwc *RWC) Close() error {
+	return nil
+}
+
+func TestJSONRequestParsing(t *testing.T) {
+	server := NewServer()
+	service := new(Service)
+
+	if err := server.RegisterName("calc", service); err != nil {
+		t.Fatalf("%v", err)
+	}
+
+	req := bytes.NewBufferString(`{"id": 1234, "jsonrpc": "2.0", "method": "calc_Add", "params": [11, 22]}`)
+	var str string
+	reply := bytes.NewBufferString(str)
+	rw := &RWC{bufio.NewReadWriter(bufio.NewReader(req), bufio.NewWriter(reply))}
+
+	codec := NewJSONCodec(rw)
+
+	requests, batch, err := codec.ReadRequestHeaders()
+	if err != nil {
+		t.Fatalf("%v", err)
+	}
+
+	if batch {
+		t.Fatalf("Request isn't a batch")
+	}
+
+	if len(requests) != 1 {
+		t.Fatalf("Expected 1 request but got %d requests - %v", len(requests), requests)
+	}
+
+	if requests[0].service != "calc" {
+		t.Fatalf("Expected service 'calc' but got '%s'", requests[0].service)
+	}
+
+	if requests[0].method != "Add" {
+		t.Fatalf("Expected method 'Add' but got '%s'", requests[0].method)
+	}
+
+	if requests[0].id != 1234 {
+		t.Fatalf("Expected id 1234 but got %d", requests[0].id)
+	}
+
+	var arg int
+	args := []reflect.Type{reflect.TypeOf(arg), reflect.TypeOf(arg)}
+
+	v, err := codec.ParseRequestArguments(args, requests[0].params)
+	if err != nil {
+		t.Fatalf("%v", err)
+	}
+
+	if len(v) != 2 {
+		t.Fatalf("Expected 2 argument values, got %d", len(v))
+	}
+
+	if v[0].Int() != 11 || v[1].Int() != 22 {
+		t.Fatalf("expected %d == 11 && %d == 22", v[0].Int(), v[1].Int())
+	}
+}

rpc/v2/server.go

@@ -0,0 +1,449 @@
+// 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 v2
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+)
+
+// NewServer will create a new server instance with no registered handlers.
+func NewServer() *Server {
+	return &Server{services: make(serviceRegistry), subscriptions: make(subscriptionRegistry)}
+}
+
+// Register publishes suitable methods of rcvr. Methods will be published
+// with a dot between the rcrv and method, e.g. Calculator.Add. The supplied
+// rcvr must be a pointer.
+func (s *Server) Register(rcvr interface{}) error {
+	return s.register(rcvr, "", false)
+}
+
+// RegisterName publishes suitable methods of rcvr. Methods will be published
+// with a dot between the given name and method, e.g. Calculator.Add. The
+// supplied rcvr must be a pointer.
+func (s *Server) RegisterName(name string, rcvr interface{}) error {
+	return s.register(rcvr, name, true)
+}
+
+// register will create an service for the given rcvr type under the given name if specified in useName.
+// When no methods on the given rcvr match the criteria to be either a RPC method or a subscription an error is
+// returned. Otherwise a new service is created and added to the service collection this server instance serves.
+func (s *Server) register(rcvr interface{}, name string, useName bool) error {
+	if s.services == nil {
+		s.services = make(serviceRegistry)
+	}
+
+	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.services[sname]; present {
+		return fmt.Errorf("%s already registered", sname)
+	}
+
+	svc.name = sname
+	svc.callbacks, svc.subscriptions = suitableCallbacks(svc.typ)
+
+	if len(svc.callbacks) == 0 && len(svc.subscriptions) == 0 {
+		return fmt.Errorf("Service doesn't have any suitable methods/subscriptions to expose")
+	}
+
+	s.services[svc.name] = svc
+
+	return nil
+}
+
+// ServeCodec reads incoming requests from codec, calls the appropriate callback and writes the
+// response back using the given codec. It will block until the codec is closed.
+//
+// This server will:
+// 1. allow for asynchronous and parallel request execution
+// 2. supports notifications (pub/sub)
+// 3. supports request batches
+func (s *Server) ServeCodec(codec ServerCodec) {
+	defer codec.Close()
+
+	for {
+		reqs, batch, err := s.readRequest(codec)
+		if err != nil {
+			glog.V(logger.Debug).Infof("%v\n", err)
+			codec.Write(codec.CreateErrorResponse(nil, err))
+			break
+		}
+
+		if batch {
+			go s.execBatch(codec, reqs)
+		} else {
+			go s.exec(codec, reqs[0])
+		}
+	}
+}
+
+// sendNotification will create a notification from the given event by serializing member fields of the event.
+// It will then send the notification to the client, when it fails the codec is closed. When the event has multiple
+// fields an array of values is returned.
+func sendNotification(codec ServerCodec, subid string, event interface{}) {
+	typ := reflect.TypeOf(event)
+	val := reflect.ValueOf(event)
+	for typ.Kind() == reflect.Ptr {
+		typ = typ.Elem()
+		val = val.Elem()
+	}
+
+	fields := make(map[string]interface{})
+	var notification interface{}
+
+	if typ.Kind() == reflect.Struct {
+		if typ.NumField() == 1 {
+			notification = codec.CreateNotification(subid, val.Field(0).Interface())
+		} else {
+			for i := 0; i < typ.NumField(); i++ {
+				fields[typ.Field(i).Name] = val.Field(i).Interface()
+			}
+			notification = codec.CreateNotification(subid, fields)
+		}
+	} else {
+		notification = codec.CreateNotification(subid, val.Interface())
+	}
+
+	if err := codec.Write(notification); err != nil {
+		codec.Close()
+	}
+}
+
+// createSubscription will register a new subscription and waits for raised events. When an event is raised it will:
+// 1. test if the event is raised matches the criteria the user has (optionally) specified
+// 2. create a notification of the event and send it the client when it matches the criteria
+// It will unsubscribe the subscription when the socket is closed or the subscription is unsubscribed by the user.
+func (s *Server) createSubscription(c ServerCodec, req *serverRequest) (string, error) {
+	args := []reflect.Value{req.rcvr}
+	if len(req.args) > 0 {
+		args = append(args, req.args...)
+	}
+
+	subid, err := newSubscriptionId()
+	if err != nil {
+		return "", err
+	}
+
+	reply := req.callb.method.Func.Call(args)
+
+	if reply[1].IsNil() { // no error
+		if subscription, ok := reply[0].Interface().(Subscription); ok {
+			s.muSubcriptions.Lock()
+			s.subscriptions[subid] = subscription
+			s.muSubcriptions.Unlock()
+			go func() {
+				cases := []reflect.SelectCase{
+					reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(subscription.Chan())}, // new event
+					reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(c.Closed())},          // connection closed
+				}
+
+				for {
+					idx, notification, recvOk := reflect.Select(cases)
+					switch idx {
+					case 0: // new event, or channel closed
+						if recvOk { // send notification
+							if event, ok := notification.Interface().(*event.Event); ok {
+								if subscription.match == nil || subscription.match(event.Data) {
+									sendNotification(c, subid, event.Data)
+								}
+							}
+						} else { // user send an eth_unsubscribe request
+							return
+						}
+					case 1: // connection closed
+						s.unsubscribe(subid)
+						return
+					}
+				}
+			}()
+		} else { // unable to create subscription
+			s.muSubcriptions.Lock()
+			delete(s.subscriptions, subid)
+			s.muSubcriptions.Unlock()
+		}
+	} else {
+		return "", fmt.Errorf("Unable to create subscription")
+	}
+
+	return subid, nil
+}
+
+// unsubscribe calls the Unsubscribe method on the subscription and removes a subscription from the subscription
+// registry.
+func (s *Server) unsubscribe(subid string) bool {
+	s.muSubcriptions.Lock()
+	defer s.muSubcriptions.Unlock()
+	if sub, ok := s.subscriptions[subid]; ok {
+		sub.Unsubscribe()
+		delete(s.subscriptions, subid)
+		return true
+	}
+	return false
+}
+
+// exec executes the given request and writes the result back using the codec.
+func (s *Server) exec(codec ServerCodec, req *serverRequest) {
+	if req.err != nil { // error during request parsing
+		rpcErr := codec.CreateErrorResponse(&req.id, req.err)
+		if err := codec.Write(rpcErr); err != nil {
+			codec.Close()
+		}
+		return
+	}
+
+	if req.isUnsubscribe { // first param must be the subscription id
+		if len(req.args) >= 1 && req.args[0].Kind() == reflect.String {
+			subid := req.args[0].String()
+			if s.unsubscribe(subid) {
+				if err := codec.Write(codec.CreateResponse(req.id, true)); err != nil {
+					codec.Close()
+				}
+			} else {
+				rpcErr := codec.CreateErrorResponse(&req.id,
+					&callbackError{fmt.Sprintf("subscription '%s' not found", subid)})
+				if err := codec.Write(rpcErr); err != nil {
+					codec.Close()
+				}
+			}
+		} else {
+			rpcErr := codec.CreateErrorResponse(&req.id, &invalidParamsError{"Expected subscription id as argument"})
+			if err := codec.Write(rpcErr); err != nil {
+				codec.Close()
+			}
+		}
+		return
+	}
+
+	if req.callb.isSubscribe {
+		subid, err := s.createSubscription(codec, req)
+		var response interface{}
+		if err == nil {
+			response = codec.CreateResponse(req.id, subid)
+		} else {
+			response = codec.CreateErrorResponse(&req.id, &callbackError{err.Error()})
+		}
+
+		if err = codec.Write(response); err != nil {
+			codec.Close()
+		}
+		return
+	}
+
+	// regular RPC call
+	if len(req.args) != len(req.callb.argTypes) {
+		rpcErr := &invalidParamsError{fmt.Sprintf("%s%s%s expects %d parameters, got %d",
+			req.svcname, serviceMethodSeparator, req.callb.method.Name,
+			len(req.callb.argTypes), len(req.args))}
+
+		res := codec.CreateErrorResponse(&req.id, rpcErr)
+		if err := codec.Write(res); err != nil {
+			codec.Close()
+		}
+		return
+	}
+
+	arguments := []reflect.Value{req.rcvr}
+	if len(req.args) > 0 {
+		arguments = append(arguments, req.args...)
+	}
+
+	reply := req.callb.method.Func.Call(arguments)
+
+	if len(reply) == 0 {
+		if err := codec.Write(codec.CreateResponse(req.id, nil)); err != nil {
+			codec.Close()
+		}
+		return
+	}
+
+	if req.callb.errPos >= 0 { // test if method returned an error
+		if !reply[req.callb.errPos].IsNil() {
+			e := reply[req.callb.errPos].Interface().(error)
+			res := codec.CreateErrorResponse(&req.id, &callbackError{e.Error()})
+			if err := codec.Write(res); err != nil {
+				codec.Close()
+			}
+			return
+		}
+	}
+
+	if err := codec.Write(codec.CreateResponse(req.id, reply[0].Interface())); err != nil {
+		codec.Close()
+	}
+}
+
+// execBatch executes the given requests and writes the result back using the codec. It will only write the response
+// back when the last request is processed.
+func (s *Server) execBatch(codec ServerCodec, requests []*serverRequest) {
+	responses := make([]interface{}, len(requests))
+
+	for i, req := range requests {
+		if req.err != nil { // error during parsing of request
+			responses[i] = codec.CreateErrorResponse(&req.id, req.err)
+			continue
+		}
+
+		if req.isUnsubscribe {
+			if len(req.args) == 1 && req.args[0].Kind() == reflect.String {
+				subid := req.args[0].String()
+				if s.unsubscribe(subid) {
+					responses[i] = codec.CreateResponse(req.id, true)
+				} else {
+					e := &callbackError{fmt.Sprintf("subscription '%s' not found", subid)}
+					responses[i] = codec.CreateErrorResponse(&req.id, e)
+				}
+			} else {
+				e := &invalidParamsError{"Expected subscription id as argument"}
+				responses[i] = codec.CreateErrorResponse(&req.id, e)
+			}
+			continue
+		}
+
+		if req.callb.isSubscribe {
+			subid, err := s.createSubscription(codec, req)
+			var response interface{}
+			if err == nil {
+				response = codec.CreateResponse(req.id, subid)
+			} else {
+				response = codec.CreateErrorResponse(&req.id, &callbackError{err.Error()})
+			}
+
+			responses[i] = response
+			continue
+		}
+
+		var reply []reflect.Value
+
+		if len(req.args) != len(req.callb.argTypes) {
+			rpcErr := &invalidParamsError{fmt.Sprintf("%s%s%s expects %d parameters, got %d",
+				req.svcname, serviceMethodSeparator, req.callb.method.Name, len(req.callb.argTypes), len(req.args))}
+			responses[i] = codec.CreateErrorResponse(&req.id, rpcErr)
+			continue
+		}
+
+		arguments := []reflect.Value{req.rcvr}
+		if len(req.args) > 0 {
+			arguments = append(arguments, req.args...)
+		}
+
+		reply = req.callb.method.Func.Call(arguments)
+
+		if len(reply) == 0 {
+			responses[i] = codec.CreateResponse(req.id, nil)
+			continue
+		}
+
+		if req.callb.errPos >= 0 {
+			if !reply[req.callb.errPos].IsNil() {
+				if e, ok := reply[req.callb.errPos].Interface().(error); ok {
+					rpcErr := &callbackError{e.Error()}
+					responses[i] = codec.CreateErrorResponse(&req.id, rpcErr)
+					continue
+				}
+			}
+		}
+
+		responses[i] = codec.CreateResponse(req.id, reply[0].Interface())
+	}
+
+	if err := codec.Write(responses); err != nil {
+		glog.V(logger.Error).Infof("%v\n", err)
+		codec.Close()
+	}
+}
+
+// readRequest requests the next (batch) request from the codec. It will return the collection of requests, an
+// indication if the request was a batch, the invalid request identifier and an error when the request could not be
+// read/parsed.
+func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, RPCError) {
+	reqs, batch, err := codec.ReadRequestHeaders()
+	if err != nil {
+		return nil, batch, err
+	}
+
+	requests := make([]*serverRequest, len(reqs))
+
+	// verify requests
+	for i, r := range reqs {
+		var ok bool
+		var svc *service
+
+		if r.isPubSub && r.method == unsubscribeMethod {
+			requests[i] = &serverRequest{id: r.id, isUnsubscribe: true}
+			argTypes := []reflect.Type{reflect.TypeOf("")}
+			if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil {
+				requests[i].args = args
+			} else {
+				requests[i].err = &invalidParamsError{err.Error()}
+			}
+			continue
+		}
+
+		if svc, ok = s.services[r.service]; !ok {
+			requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}}
+			continue
+		}
+
+		if callb, ok := svc.subscriptions[r.method]; ok {
+			requests[i] = &serverRequest{id: r.id, svcname: svc.name, rcvr: svc.rcvr, callb: callb}
+			if r.params != nil && len(callb.argTypes) > 0 {
+				argTypes := []reflect.Type{reflect.TypeOf("")}
+				argTypes = append(argTypes, callb.argTypes...)
+				if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil {
+					requests[i].args = args[1:] // first one is service.method name which isn't an actual argument
+				} else {
+					requests[i].err = &invalidParamsError{err.Error()}
+				}
+			}
+			continue
+		}
+
+		if callb, ok := svc.callbacks[r.method]; ok {
+			requests[i] = &serverRequest{id: r.id, svcname: svc.name, rcvr: svc.rcvr, callb: callb}
+			if r.params != nil && len(callb.argTypes) > 0 {
+				if args, err := codec.ParseRequestArguments(callb.argTypes, r.params); err == nil {
+					requests[i].args = args
+				} else {
+					requests[i].err = &invalidParamsError{err.Error()}
+				}
+			}
+			continue
+		}
+
+		requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}}
+	}
+
+	return requests, batch, nil
+}

rpc/v2/server_test.go

@@ -0,0 +1,57 @@
+package v2
+
+import "testing"
+
+type Service struct{}
+
+type Args struct {
+	S string
+}
+
+func (s *Service) NoArgsRets() {
+}
+
+func (s *Service) Args(str string, i int, args *Args) {
+}
+
+func (s *Service) Rets() (string, error) {
+	return "", nil
+}
+
+func (s *Service) InvalidRets1() (error, string) {
+	return nil, ""
+}
+
+func (s *Service) InvalidRets2() (string, string) {
+	return "", ""
+}
+
+func (s *Service) InvalidRets3() (string, string, error) {
+	return "", "", nil
+}
+
+func TestServerRegister(t *testing.T) {
+	server := NewServer()
+	service := new(Service)
+
+	if err := server.RegisterName("calc", service); err != nil {
+		t.Fatalf("%v", err)
+	}
+
+	if err := server.RegisterName("calc", service); err == nil {
+		t.Fatal("Second time registering the same service should raise an error")
+	}
+
+	if len(server.services) != 1 {
+		t.Fatalf("Expected 1 service entry but got %d", len(server.services))
+	}
+
+	svc, ok := server.services["calc"]
+	if !ok {
+		t.Fatalf("Expected service calc to be registered")
+	}
+
+	if len(svc.callbacks) != 3 {
+		t.Fatalf("Expected 3 callbacks for service 'calc', got %d", len(svc.callbacks))
+	}
+}

rpc/v2/types.go

@@ -0,0 +1,137 @@
+// 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 v2
+
+import (
+	"reflect"
+
+	"sync"
+
+	"github.com/ethereum/go-ethereum/event"
+)
+
+// callback is a method callback which was registered in the server
+type callback struct {
+	method      reflect.Method // callback
+	argTypes    []reflect.Type // input argument types
+	errPos      int            // err return idx, of -1 when method cannot return error
+	isSubscribe bool           // indication if the callback is a subscription
+}
+
+// service represents a registered object
+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
+	subscriptions subscriptions // available subscriptions/notifications
+}
+
+// serverRequest is an incoming request
+type serverRequest struct {
+	id            int64
+	svcname       string
+	rcvr          reflect.Value
+	callb         *callback
+	args          []reflect.Value
+	isUnsubscribe bool
+	err           RPCError
+}
+
+type serviceRegistry map[string]*service          // collection of services
+type callbacks map[string]*callback               // collection of RPC callbacks
+type subscriptions map[string]*callback           // collection of subscription callbacks
+type subscriptionRegistry map[string]Subscription // collection of subscriptions
+
+// Server represents a RPC server
+type Server struct {
+	services       serviceRegistry
+	muSubcriptions sync.Mutex // protects subscriptions
+	subscriptions  subscriptionRegistry
+}
+
+// rpcRequest represents a raw incoming RPC request
+type rpcRequest struct {
+	service  string
+	method   string
+	id       int64
+	isPubSub bool
+	params   interface{}
+}
+
+// RPCError implements RPC error, is add support for error codec over regular go errors
+type RPCError interface {
+	// RPC error code
+	Code() int
+	// Error message
+	Error() string
+}
+
+// ServerCodec implements reading, parsing and writing RPC messages for the server side of
+// a RPC session. Implementations must be go-routine safe since the codec can be called in
+// multiple go-routines concurrently.
+type ServerCodec interface {
+	// Read next request
+	ReadRequestHeaders() ([]rpcRequest, bool, RPCError)
+	// Parse request argument to the given types
+	ParseRequestArguments([]reflect.Type, interface{}) ([]reflect.Value, RPCError)
+	// Assemble success response
+	CreateResponse(int64, interface{}) interface{}
+	// Assemble error response
+	CreateErrorResponse(*int64, RPCError) interface{}
+	// Assemble error response with extra information about the error through info
+	CreateErrorResponseWithInfo(id *int64, err RPCError, info interface{}) interface{}
+	// Create notification response
+	CreateNotification(string, interface{}) interface{}
+	// Write msg to client.
+	Write(interface{}) error
+	// Close underlying data stream
+	Close()
+	// Closed when underlying connection is closed
+	Closed() <-chan interface{}
+}
+
+// SubscriptionMatcher returns true if the given value matches the criteria specified by the user
+type SubscriptionMatcher func(interface{}) bool
+
+// Subscription is used by the server to send notifications to the client
+type Subscription struct {
+	sub   event.Subscription
+	match SubscriptionMatcher
+}
+
+// NewSubscription create a new RPC subscription
+func NewSubscription(sub event.Subscription) Subscription {
+	return Subscription{sub, nil}
+}
+
+// NewSubscriptionFiltered will create a new subscription. For each raised event the given matcher is
+// called. If it returns true the event is send as notification to the client, otherwise it is ignored.
+func NewSubscriptionFiltered(sub event.Subscription, match SubscriptionMatcher) Subscription {
+	return Subscription{sub, match}
+}
+
+// Chan returns the channel where new events will be published. It's up the user to call the matcher to
+// determine if the events are interesting for the client.
+func (s *Subscription) Chan() <-chan *event.Event {
+	return s.sub.Chan()
+}
+
+// Unsubscribe will end the subscription and closes the event channel
+func (s *Subscription) Unsubscribe() {
+	s.sub.Unsubscribe()
+}

rpc/v2/utils.go

@@ -0,0 +1,169 @@
+// 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 v2
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"reflect"
+	"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)
+}
+
+var subscriptionType = reflect.TypeOf((*Subscription)(nil)).Elem()
+
+func isSubscriptionType(t reflect.Type) bool {
+	for t.Kind() == reflect.Ptr {
+		t = t.Elem()
+	}
+	return t == subscriptionType
+}
+
+// isPubSub tests whether the given method return the pair
+// (v2.Subscription, error)
+func isPubSub(methodType reflect.Type) bool {
+	if methodType.NumOut() != 2 {
+		return false
+	}
+
+	return isSubscriptionType(methodType.Out(0)) && isErrorType(methodType.Out(1))
+}
+
+// suitableCallbacks iterates over the methods of the given type. It will determine if a method satisfies the criteria
+// for a RPC callback or a subscription callback and adds it to the collection of callbacks or subscriptions. See server
+// documentation for a summary of these criteria.
+func suitableCallbacks(typ reflect.Type) (callbacks, subscriptions) {
+	callbacks := make(callbacks)
+	subscriptions := make(subscriptions)
+
+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
+		}
+
+		if isPubSub(mtype) {
+			var h callback
+			h.method = method
+			h.errPos = -1
+			h.isSubscribe = true
+
+			h.argTypes = make([]reflect.Type, mtype.NumIn()-1) // skip rcvr type
+			for i := 1; i < mtype.NumIn(); i++ {
+				argType := mtype.In(i)
+				if isExportedOrBuiltinType(argType) {
+					h.argTypes[i-1] = argType
+				} else {
+					continue METHODS
+				}
+			}
+
+			subscriptions[mname] = &h
+			continue METHODS
+		}
+
+		var h callback
+		h.method = method
+		numIn := mtype.NumIn()
+
+		// determine method arguments, ignore first arg since it's the receiver type
+		// Arguments must be exported or builtin
+		h.argTypes = make([]reflect.Type, numIn-1)
+		for i := 1; i < numIn; i++ {
+			argType := mtype.In(i)
+			if !isExportedOrBuiltinType(argType) {
+				continue METHODS
+			}
+			h.argTypes[i-1] = argType
+		}
+
+		// check that all returned values are exported or builtin values
+		for i := 0; i < mtype.NumOut(); i++ {
+			if !isExportedOrBuiltinType(mtype.Out(i)) {
+				continue METHODS
+			}
+		}
+
+		// when a method returns an error it must be the last returned value
+		h.errPos = -1
+		for i := 0; i < mtype.NumOut(); i++ {
+			if isErrorType(mtype.Out(i)) {
+				h.errPos = i
+				break
+			}
+		}
+
+		if h.errPos >= 0 && h.errPos != mtype.NumOut()-1 {
+			continue METHODS
+		}
+
+		switch mtype.NumOut() {
+		case 0, 1:
+			break
+		case 2:
+			if h.errPos == -1 { // method must one return value and 1 error
+				continue METHODS
+			}
+			break
+		default:
+			continue METHODS
+		}
+
+		callbacks[mname] = &h
+	}
+
+	return callbacks, subscriptions
+}
+
+func newSubscriptionId() (string, error) {
+	var subid [16]byte
+	n, _ := rand.Read(subid[:])
+	if n != 16 {
+		return "", errors.New("Unable to generate subscription id")
+	}
+	return "0x" + hex.EncodeToString(subid[:]), nil
+}