aboutsummaryrefslogtreecommitdiffstats
path: root/rpc/server.go
diff options
context:
space:
mode:
Diffstat (limited to 'rpc/server.go')
-rw-r--r--rpc/server.go438
1 files changed, 438 insertions, 0 deletions
diff --git a/rpc/server.go b/rpc/server.go
new file mode 100644
index 000000000..0b93a4e64
--- /dev/null
+++ b/rpc/server.go
@@ -0,0 +1,438 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package rpc
+
+import (
+ "fmt"
+ "reflect"
+ "runtime"
+ "sync/atomic"
+ "time"
+
+ "github.com/ethereum/go-ethereum/event"
+ "github.com/ethereum/go-ethereum/logger"
+ "github.com/ethereum/go-ethereum/logger/glog"
+ "golang.org/x/net/context"
+ "gopkg.in/fatih/set.v0"
+)
+
+const (
+ stopPendingRequestTimeout = 3 * time.Second // give pending requests stopPendingRequestTimeout the time to finish when the server is stopped
+
+ DefaultIpcApis = "admin,eth,debug,miner,net,shh,txpool,personal,web3"
+ DefaultHttpRpcApis = "eth,net,web3"
+)
+
+// NewServer will create a new server instance with no registered handlers.
+func NewServer() *Server {
+ server := &Server{
+ services: make(serviceRegistry),
+ subscriptions: make(subscriptionRegistry),
+ codecs: set.New(),
+ run: 1,
+ }
+
+ // 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()
+
+ s.codecsMu.Lock()
+ if atomic.LoadInt32(&s.run) != 1 { // server stopped
+ s.codecsMu.Unlock()
+ return
+ }
+ s.codecs.Add(codec)
+ s.codecsMu.Unlock()
+
+ for atomic.LoadInt32(&s.run) == 1 {
+ 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 atomic.LoadInt32(&s.run) != 1 {
+ err = &shutdownError{}
+ if batch {
+ resps := make([]interface{}, len(reqs))
+ for i, r := range reqs {
+ resps[i] = codec.CreateErrorResponse(&r.id, err)
+ }
+ codec.Write(resps)
+ } else {
+ codec.Write(codec.CreateErrorResponse(&reqs[0].id, err))
+ }
+ break
+ }
+
+ if batch {
+ go s.execBatch(ctx, codec, reqs)
+ } else {
+ go s.exec(ctx, codec, reqs[0])
+ }
+ }
+}
+
+// Stop will stop reading new requests, wait for stopPendingRequestTimeout to allow pending requests to finish,
+// close all codecs which will cancels pending requests/subscriptions.
+func (s *Server) Stop() {
+ if atomic.CompareAndSwapInt32(&s.run, 1, 0) {
+ glog.V(logger.Debug).Infoln("RPC Server shutdown initiatied")
+ time.AfterFunc(stopPendingRequestTimeout, func() {
+ s.codecsMu.Lock()
+ defer s.codecsMu.Unlock()
+ s.codecs.Each(func(c interface{}) bool {
+ c.(ServerCodec).Close()
+ return true
+ })
+ })
+ }
+}
+
+// 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
+}