aboutsummaryrefslogtreecommitdiffstats
path: root/rpc/v2/server.go
diff options
context:
space:
mode:
authorBas van Kervel <bas@ethdev.com>2015-12-16 17:58:01 +0800
committerJeffrey Wilcke <geffobscura@gmail.com>2016-01-26 20:51:50 +0800
commit19b2640e89465c1c57f1bbea0274d52d97151f60 (patch)
tree980e063693dae7fa6105646821ee6755b176b6e2 /rpc/v2/server.go
parentf2ab351e8d3b0a4e569ce56f6a4f17725ca5ba65 (diff)
downloadgo-tangerine-19b2640e89465c1c57f1bbea0274d52d97151f60.tar.gz
go-tangerine-19b2640e89465c1c57f1bbea0274d52d97151f60.tar.zst
go-tangerine-19b2640e89465c1c57f1bbea0274d52d97151f60.zip
rpc: migrated the RPC insterface to a new reflection based RPC layer
Diffstat (limited to 'rpc/v2/server.go')
-rw-r--r--rpc/v2/server.go385
1 files changed, 0 insertions, 385 deletions
diff --git a/rpc/v2/server.go b/rpc/v2/server.go
deleted file mode 100644
index 4c04f04d2..000000000
--- a/rpc/v2/server.go
+++ /dev/null
@@ -1,385 +0,0 @@
-// 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"
-
- "runtime"
-
- "github.com/ethereum/go-ethereum/event"
- "github.com/ethereum/go-ethereum/logger"
- "github.com/ethereum/go-ethereum/logger/glog"
- "golang.org/x/net/context"
-)
-
-// NewServer will create a new server instance with no registered handlers.
-func NewServer() *Server {
- server := &Server{services: make(serviceRegistry), subscriptions: make(subscriptionRegistry)}
-
- // register a default service which will provide meta information about the RPC service such as the services and
- // methods it offers.
- rpcService := &RPCService{server}
- server.RegisterName("rpc", rpcService)
-
- return server
-}
-
-// RPCService gives meta information about the server.
-// e.g. gives information about the loaded modules.
-type RPCService struct {
- server *Server
-}
-
-// Modules returns the list of RPC services with their version number
-func (s *RPCService) Modules() map[string]string {
- modules := make(map[string]string)
- for name, _ := range s.server.services {
- modules[name] = "1.0"
- }
- return modules
-}
-
-// RegisterName will create an service for the given rcvr type under the given name. 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) RegisterName(name string, rcvr interface{}) error {
- if s.services == nil {
- s.services = make(serviceRegistry)
- }
-
- svc := new(service)
- svc.typ = reflect.TypeOf(rcvr)
- rcvrVal := reflect.ValueOf(rcvr)
-
- if name == "" {
- return fmt.Errorf("no service name for type %s", svc.typ.String())
- }
- if !isExported(reflect.Indirect(rcvrVal).Type().Name()) {
- return fmt.Errorf("%s is not exported", reflect.Indirect(rcvrVal).Type().Name())
- }
-
- // already a previous service register under given sname, merge methods/subscriptions
- if regsvc, present := s.services[name]; present {
- methods, subscriptions := suitableCallbacks(rcvrVal, svc.typ)
- if len(methods) == 0 && len(subscriptions) == 0 {
- return fmt.Errorf("Service doesn't have any suitable methods/subscriptions to expose")
- }
-
- for _, m := range methods {
- regsvc.callbacks[formatName(m.method.Name)] = m
- }
- for _, s := range subscriptions {
- regsvc.subscriptions[formatName(s.method.Name)] = s
- }
-
- return nil
- }
-
- svc.name = name
- svc.callbacks, svc.subscriptions = suitableCallbacks(rcvrVal, 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 func() {
- if err := recover(); err != nil {
- const size = 64 << 10
- buf := make([]byte, size)
- buf = buf[:runtime.Stack(buf, false)]
- glog.Errorln(string(buf))
- }
- codec.Close()
- }()
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
-
- 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(ctx, codec, reqs)
- } else {
- go s.exec(ctx, 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{}) {
- notification := codec.CreateNotification(subid, event)
-
- 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.callb.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, subscription.format(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
-}
-
-// handle executes a request and returns the response from the callback.
-func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverRequest) interface{} {
- if req.err != nil {
- return codec.CreateErrorResponse(&req.id, req.err)
- }
-
- 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) {
- return codec.CreateResponse(req.id, true)
- } else {
- return codec.CreateErrorResponse(&req.id,
- &callbackError{fmt.Sprintf("subscription '%s' not found", subid)})
- }
- }
- return codec.CreateErrorResponse(&req.id, &invalidParamsError{"Expected subscription id as argument"})
- }
-
- if req.callb.isSubscribe {
- subid, err := s.createSubscription(codec, req)
- if err != nil {
- return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()})
- }
- return codec.CreateResponse(req.id, subid)
- }
-
- // 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))}
- return codec.CreateErrorResponse(&req.id, rpcErr)
- }
-
- arguments := []reflect.Value{req.callb.rcvr}
- if req.callb.hasCtx {
- arguments = append(arguments, reflect.ValueOf(ctx))
- }
- if len(req.args) > 0 {
- arguments = append(arguments, req.args...)
- }
-
- reply := req.callb.method.Func.Call(arguments)
-
- if len(reply) == 0 {
- return codec.CreateResponse(req.id, nil)
- }
-
- 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()})
- return res
- }
- }
-
- return codec.CreateResponse(req.id, reply[0].Interface())
-}
-
-// exec executes the given request and writes the result back using the codec.
-func (s *Server) exec(ctx context.Context, codec ServerCodec, req *serverRequest) {
- var response interface{}
- if req.err != nil {
- response = codec.CreateErrorResponse(&req.id, req.err)
- } else {
- response = s.handle(ctx, codec, req)
- }
-
- if err := codec.Write(response); err != nil {
- glog.V(logger.Error).Infof("%v\n", err)
- 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(ctx context.Context, codec ServerCodec, requests []*serverRequest) {
- responses := make([]interface{}, len(requests))
- for i, req := range requests {
- if req.err != nil {
- responses[i] = codec.CreateErrorResponse(&req.id, req.err)
- } else {
- responses[i] = s.handle(ctx, codec, req)
- }
- }
-
- 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 r.isPubSub { // eth_subscribe
- if callb, ok := svc.subscriptions[r.method]; ok {
- requests[i] = &serverRequest{id: r.id, svcname: svc.name, 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()}
- }
- }
- } else {
- requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{subscribeMethod, r.method}}
- }
- continue
- }
-
- if callb, ok := svc.callbacks[r.method]; ok {
- requests[i] = &serverRequest{id: r.id, svcname: svc.name, 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
-}