added comms http

10 years ago · fd764d4ff7
parent 60c2ccd99c
commit fd764d4ff7
5 changed files with 254 additions and 1 deletions
--- a/rpc/api/mergedapi.go
+++ b/rpc/api/mergedapi.go
@ -1,6 +1,8 @@
 package api

 import (
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/rpc/shared"
 )

@ -40,6 +42,8 @@ func (self *MergedApi) Methods() []string {

 // Call the correct API's Execute method for the given request
 func (self *MergedApi) Execute(req *shared.Request) (interface{}, error) {
+	glog.V(logger.Detail).Infof("rpc method: %s", req.Method)
+
 	if res, _ := self.handle(req); res != nil {
 		return res, nil
 	}
--- a/rpc/comms/comms.go
+++ b/rpc/comms/comms.go
@ -11,6 +11,11 @@ import (
 	"github.com/ethereum/go-ethereum/rpc/shared"
 )

+const (
+	jsonrpcver           = "2.0"
+	maxHttpSizeReqLength = 1024 * 1024 // 1MB
+)
+
 type EthereumClient interface {
 	Close()
 	Send(interface{}) error
@ -26,7 +31,7 @@ func handle(conn net.Conn, api api.EthereumApi, c codec.Codec) {
 			codec.Close()
 			return
 		} else if err != nil {
-			glog.V(logger.Error).Infof("IPC recv err - %v\n", err)
+			glog.V(logger.Error).Infof("comms recv err - %v\n", err)
 			codec.Close()
 			return
 		}
--- a/rpc/comms/http.go
+++ b/rpc/comms/http.go
@ -0,0 +1,65 @@
+package comms
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/rpc/api"
+	"github.com/ethereum/go-ethereum/rpc/codec"
+	"github.com/rs/cors"
+)
+
+var (
+	// main HTTP rpc listener
+	httpListener         *stoppableTCPListener
+	listenerStoppedError = fmt.Errorf("Listener has stopped")
+)
+
+type HttpConfig struct {
+	ListenAddress string
+	ListenPort    uint
+	CorsDomain    string
+}
+
+func StartHttp(cfg HttpConfig, codec codec.Codec, apis ...api.EthereumApi) error {
+	if httpListener != nil {
+		if fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort) != httpListener.Addr().String() {
+			return fmt.Errorf("RPC service already running on %s ", httpListener.Addr().String())
+		}
+		return nil // RPC service already running on given host/port
+	}
+
+	l, err := newStoppableTCPListener(fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort))
+	if err != nil {
+		glog.V(logger.Error).Infof("Can't listen on %s:%d: %v", cfg.ListenAddress, cfg.ListenPort, err)
+		return err
+	}
+	httpListener = l
+
+	api := api.Merge(apis...)
+	var handler http.Handler
+	if len(cfg.CorsDomain) > 0 {
+		var opts cors.Options
+		opts.AllowedMethods = []string{"POST"}
+		opts.AllowedOrigins = strings.Split(cfg.CorsDomain, " ")
+
+		c := cors.New(opts)
+		handler = newStoppableHandler(c.Handler(gethHttpHandler(codec, api)), l.stop)
+	} else {
+		handler = newStoppableHandler(gethHttpHandler(codec, api), l.stop)
+	}
+
+	go http.Serve(l, handler)
+
+	return nil
+}
+
+func StopHttp() {
+	if httpListener != nil {
+		httpListener.Stop()
+		httpListener = nil
+	}
+}
--- a/rpc/comms/http_net.go
+++ b/rpc/comms/http_net.go
@ -0,0 +1,167 @@
+package comms
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/rpc/api"
+	"github.com/ethereum/go-ethereum/rpc/codec"
+	"github.com/ethereum/go-ethereum/rpc/shared"
+)
+
+// When https://github.com/golang/go/issues/4674 is implemented this could be replaced
+type stoppableTCPListener struct {
+	*net.TCPListener
+	stop chan struct{} // closed when the listener must stop
+}
+
+func newStoppableTCPListener(addr string) (*stoppableTCPListener, error) {
+	wl, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	if tcpl, ok := wl.(*net.TCPListener); ok {
+		stop := make(chan struct{})
+		return &stoppableTCPListener{tcpl, stop}, nil
+	}
+
+	return nil, fmt.Errorf("Unable to create TCP listener for RPC service")
+}
+
+// Stop the listener and all accepted and still active connections.
+func (self *stoppableTCPListener) Stop() {
+	close(self.stop)
+}
+
+func (self *stoppableTCPListener) Accept() (net.Conn, error) {
+	for {
+		self.SetDeadline(time.Now().Add(time.Duration(1 * time.Second)))
+		c, err := self.TCPListener.AcceptTCP()
+
+		select {
+		case <-self.stop:
+			if c != nil { // accept timeout
+				c.Close()
+			}
+			self.TCPListener.Close()
+			return nil, listenerStoppedError
+		default:
+		}
+
+		if err != nil {
+			if netErr, ok := err.(net.Error); ok && netErr.Timeout() && netErr.Temporary() {
+				continue // regular timeout
+			}
+		}
+
+		return &closableConnection{c, self.stop}, err
+	}
+}
+
+type closableConnection struct {
+	*net.TCPConn
+	closed chan struct{}
+}
+
+func (self *closableConnection) Read(b []byte) (n int, err error) {
+	select {
+	case <-self.closed:
+		self.TCPConn.Close()
+		return 0, io.EOF
+	default:
+		return self.TCPConn.Read(b)
+	}
+}
+
+// Wraps the default handler and checks if the RPC service was stopped. In that case it returns an
+// error indicating that the service was stopped. This will only happen for connections which are
+// kept open (HTTP keep-alive) when the RPC service was shutdown.
+func newStoppableHandler(h http.Handler, stop chan struct{}) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case <-stop:
+			w.Header().Set("Content-Type", "application/json")
+			err := fmt.Errorf("RPC service stopped")
+			response := shared.NewRpcResponse(-1, jsonrpcver, nil, err)
+			httpSend(w, response)
+		default:
+			h.ServeHTTP(w, r)
+		}
+	})
+}
+
+func httpSend(writer io.Writer, v interface{}) (n int, err error) {
+	var payload []byte
+	payload, err = json.MarshalIndent(v, "", "\t")
+	if err != nil {
+		glog.V(logger.Error).Infoln("Error marshalling JSON", err)
+		return 0, err
+	}
+	glog.V(logger.Detail).Infof("Sending payload: %s", payload)
+
+	return writer.Write(payload)
+}
+
+func gethHttpHandler(codec codec.Codec, api api.EthereumApi) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+
+		// Limit request size to resist DoS
+		if req.ContentLength > maxHttpSizeReqLength {
+			err := fmt.Errorf("Request too large")
+			response := shared.NewRpcErrorResponse(-1, jsonrpcver, -32700, err)
+			httpSend(w, &response)
+			return
+		}
+
+		defer req.Body.Close()
+		payload, err := ioutil.ReadAll(req.Body)
+		if err != nil {
+			err := fmt.Errorf("Could not read request body")
+			response := shared.NewRpcErrorResponse(-1, jsonrpcver, -32700, err)
+			httpSend(w, &response)
+			return
+		}
+
+		c := codec.New(nil)
+		var rpcReq shared.Request
+		if err = c.Decode(payload, &rpcReq); err == nil {
+			reply, err := api.Execute(&rpcReq)
+			res := shared.NewRpcResponse(rpcReq.Id, rpcReq.Jsonrpc, reply, err)
+			httpSend(w, &res)
+			return
+		}
+
+		var reqBatch []shared.Request
+		if err = c.Decode(payload, &reqBatch); err == nil {
+			resBatch := make([]*interface{}, len(reqBatch))
+			resCount := 0
+
+			for i, rpcReq := range reqBatch {
+				reply, err := api.Execute(&rpcReq)
+				if rpcReq.Id != nil { // this leaves nil entries in the response batch for later removal
+					resBatch[i] = shared.NewRpcResponse(rpcReq.Id, rpcReq.Jsonrpc, reply, err)
+					resCount += 1
+				}
+			}
+
+			// make response omitting nil entries
+			resBatch = resBatch[:resCount]
+			httpSend(w, resBatch)
+			return
+		}
+
+		// invalid request
+		err = fmt.Errorf("Could not decode request")
+		res := shared.NewRpcErrorResponse(-1, jsonrpcver, -32600, err)
+		httpSend(w, res)
+	})
+}
--- a/rpc/shared/types.go
+++ b/rpc/shared/types.go
@ -42,6 +42,18 @@ type ErrorObject struct {
 	// Data    interface{} `json:"data"`
 }

+// Create RPC error response, this allows for custom error codes
+func NewRpcErrorResponse(id interface{}, jsonrpcver string, errCode int, err error) *interface{} {
+	var response interface{}
+
+	jsonerr := &ErrorObject{errCode, err.Error()}
+	response = ErrorResponse{Jsonrpc: jsonrpcver, Id: id, Error: jsonerr}
+
+	glog.V(logger.Detail).Infof("Generated error response: %s", response)
+	return &response
+}
+
+// Create RPC response
 func NewRpcResponse(id interface{}, jsonrpcver string, reply interface{}, err error) *interface{} {
 	var response interface{}