From fd764d4ff7bd7e75dfdd7f733f50dd3805d4b3f2 Mon Sep 17 00:00:00 2001
From: Bas van Kervel <bas@ethdev.com>
Date: Tue, 16 Jun 2015 13:07:13 +0200
Subject: added comms http
---
rpc/api/mergedapi.go | 4 ++
rpc/comms/comms.go | 7 ++-
rpc/comms/http.go | 65 ++++++++++++++++++++
rpc/comms/http_net.go | 167 ++++++++++++++++++++++++++++++++++++++++++++++++++
rpc/shared/types.go | 12 ++++
5 files changed, 254 insertions(+), 1 deletion(-)
create mode 100644 rpc/comms/http.go
create mode 100644 rpc/comms/http_net.go
(limited to 'rpc')
diff --git a/rpc/api/mergedapi.go b/rpc/api/mergedapi.go
index b62477a14..8cf4ae96e 100644
--- 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
}
diff --git a/rpc/comms/comms.go b/rpc/comms/comms.go
index 3f458adfd..2af63e55d 100644
--- 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
}
diff --git a/rpc/comms/http.go b/rpc/comms/http.go
new file mode 100644
index 000000000..1fea8dc1d
--- /dev/null
+++ 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
+ }
+}
diff --git a/rpc/comms/http_net.go b/rpc/comms/http_net.go
new file mode 100644
index 000000000..8d1bacc06
--- /dev/null
+++ 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)
+ })
+}
diff --git a/rpc/shared/types.go b/rpc/shared/types.go
index 6a29fa88e..dd38a4559 100644
--- 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{}
--
cgit v1.2.3