Merge pull request #2 from renaynay/handler-stack

Shifted handler stack creation to pkg node from pkg rpc

cmd/geth/retesteth.go

@@ -887,6 +887,12 @@ func retesteth(ctx *cli.Context) error {
 	}
 	vhosts := splitAndTrim(ctx.GlobalString(utils.RPCVirtualHostsFlag.Name))
 	cors := splitAndTrim(ctx.GlobalString(utils.RPCCORSDomainFlag.Name))
+	wsOrigins := splitAndTrim(ctx.GlobalString(utils.WSAllowedOriginsFlag.Value))
+
+	srv := rpc.NewServer()
+
+	handler := rpc.NewHTTPHandlerStack(srv, cors, vhosts)
+	handler = rpc.NewWebsocketUpgradeHandler(handler, srv.WebsocketHandler(wsOrigins))
 
 	// start http server
 	var RetestethHTTPTimeouts = rpc.HTTPTimeouts{
@@ -895,7 +901,7 @@ func retesteth(ctx *cli.Context) error {
 		IdleTimeout:  120 * time.Second,
 	}
 	httpEndpoint := fmt.Sprintf("%s:%d", ctx.GlobalString(utils.RPCListenAddrFlag.Name), ctx.Int(rpcPortFlag.Name))
-	listener, _, err := rpc.StartHTTPEndpoint(httpEndpoint, rpcAPI, []string{"test", "eth", "debug", "web3"}, cors, vhosts, RetestethHTTPTimeouts, []string{})
+	listener, err := rpc.StartHTTPEndpoint(httpEndpoint, rpcAPI, []string{"test", "eth", "debug", "web3"}, RetestethHTTPTimeouts, handler)
 	if err != nil {
 		utils.Fatalf("Could not start RPC api: %v", err)
 	}

node/node.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -368,7 +369,20 @@ func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors
 	if endpoint == "" {
 		return nil
 	}
-	listener, handler, err := rpc.StartHTTPEndpoint(endpoint, apis, modules, cors, vhosts, timeouts, wsOrigins)
+
+	srv := rpc.NewServer()
+
+	err := RegisterApisFromWhitelist(apis, modules, srv)
+
+	var ws http.Handler
+	if n.httpEndpoint == n.wsEndpoint {
+		ws = srv.WebsocketHandler(wsOrigins)
+	}
+
+	// wrap handler in websocket handler only if websocket port is the same as http rpc
+	handler := n.AddWebsocketHandler(rpc.NewHTTPHandlerStack(srv, cors, vhosts), ws)
+
+	listener, err := rpc.StartHTTPEndpoint(endpoint, apis, modules, timeouts, handler)
 	if err != nil {
 		return err
 	}
@@ -383,11 +397,20 @@ func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors
 	// All listeners booted successfully
 	n.httpEndpoint = endpoint
 	n.httpListener = listener
-	n.httpHandler = handler
+	n.httpHandler = srv
 
 	return nil
 }
 
+// AddWebsocketHandler creates the handler stack necessary to handle both http rpc requests and websocket requests
+func (n *Node) AddWebsocketHandler(handler http.Handler, websocket http.Handler) http.Handler {
+	if websocket != nil {
+		return rpc.NewWebsocketUpgradeHandler(handler, websocket)
+	}
+
+	return handler
+}
+
 // stopHTTP terminates the HTTP RPC endpoint.
 func (n *Node) stopHTTP() {
 	if n.httpListener != nil {
@@ -673,3 +696,23 @@ func (n *Node) apis() []rpc.API {
 		},
 	}
 }
+
+func RegisterApisFromWhitelist(apis []rpc.API, modules []string, srv *rpc.Server) error {
+	// Generate the whitelist based on the allowed modules
+	whitelist := make(map[string]bool)
+	for _, module := range modules {
+		whitelist[module] = true
+	}
+
+	// Register all the APIs exposed by the services
+	for _, api := range apis {
+		if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
+			if err := srv.RegisterName(api.Namespace, api.Service); err != nil {
+				return err
+			}
+			log.Debug("HTTP registered", "namespace", api.Namespace)
+		}
+	}
+
+	return nil
+}

rpc/endpoints.go

@@ -18,6 +18,8 @@ package rpc
 
 import (
 	"net"
+	"net/http"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 )
@@ -42,37 +44,44 @@ func checkModuleAvailability(modules []string, apis []API) (bad, available []str
 }
 
 // StartHTTPEndpoint starts the HTTP RPC endpoint, configured with cors/vhosts/modules.
-func StartHTTPEndpoint(endpoint string, apis []API, modules []string, cors []string, vhosts []string, timeouts HTTPTimeouts, wsOrigins []string) (net.Listener, *Server, error) {
+func StartHTTPEndpoint(endpoint string, apis []API, modules []string, timeouts HTTPTimeouts, handler http.Handler) (net.Listener, error) {
 	if bad, available := checkModuleAvailability(modules, apis); len(bad) > 0 {
 		log.Error("Unavailable modules in HTTP API list", "unavailable", bad, "available", available)
 	}
-	// Generate the whitelist based on the allowed modules
+
-	whitelist := make(map[string]bool)
+	// Start the HTTP listener
-	for _, module := range modules {
-		whitelist[module] = true
-	}
-	// Register all the APIs exposed by the services
-	handler := NewServer()
-	for _, api := range apis {
-		if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
-			if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
-				return nil, nil, err
-			}
-			log.Debug("HTTP registered", "namespace", api.Namespace)
-		}
-	}
-	// All APIs registered, start the HTTP listener
 	var (
 		listener net.Listener
 		err      error
 	)
 	if listener, err = net.Listen("tcp", endpoint); err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
+	// Make sure timeout values are meaningful
+	if timeouts.ReadTimeout < time.Second {
+		log.Warn("Sanitizing invalid HTTP read timeout", "provided", timeouts.ReadTimeout, "updated", DefaultHTTPTimeouts.ReadTimeout)
+		timeouts.ReadTimeout = DefaultHTTPTimeouts.ReadTimeout
+	}
+	if timeouts.WriteTimeout < time.Second {
+		log.Warn("Sanitizing invalid HTTP write timeout", "provided", timeouts.WriteTimeout, "updated", DefaultHTTPTimeouts.WriteTimeout)
+		timeouts.WriteTimeout = DefaultHTTPTimeouts.WriteTimeout
+	}
+	if timeouts.IdleTimeout < time.Second {
+		log.Warn("Sanitizing invalid HTTP idle timeout", "provided", timeouts.IdleTimeout, "updated", DefaultHTTPTimeouts.IdleTimeout)
+		timeouts.IdleTimeout = DefaultHTTPTimeouts.IdleTimeout
+	}
 
-	go NewHTTPServer(cors, vhosts, timeouts, handler, handler.WebsocketHandler(wsOrigins)).Serve(listener)
+	// Bundle and start the HTTP server
-	return listener, handler, err
+	httpSrv := &http.Server{
+		Handler:      handler,
+		ReadTimeout:  timeouts.ReadTimeout,
+		WriteTimeout: timeouts.WriteTimeout,
+		IdleTimeout:  timeouts.IdleTimeout,
+	}
+
+	go httpSrv.Serve(listener)
+	return listener, err
 }
 
 // StartWSEndpoint starts a websocket endpoint.

rpc/http.go

@@ -217,7 +217,7 @@ func NewHTTPServer(cors []string, vhosts []string, timeouts HTTPTimeouts, srv ht
 	handler := newCorsHandler(srv, cors)
 	handler = newVHostHandler(vhosts, handler)
 	handler = newGzipHandler(handler)
-	handler = newWebsocketUpgradeHandler(handler, ws)
+	handler = NewWebsocketUpgradeHandler(handler, ws)
 
 	// Make sure timeout values are meaningful
 	if timeouts.ReadTimeout < time.Second {
@@ -298,6 +298,14 @@ func validateRequest(r *http.Request) (int, error) {
 	return http.StatusUnsupportedMediaType, err
 }
 
+// NewHTTPHandlerStack returns wrapped http-related handlers
+func NewHTTPHandlerStack(srv *Server, cors []string, vhosts []string) http.Handler {
+	// Wrap the CORS-handler within a host-handler
+	handler := newCorsHandler(srv, cors)
+	handler = newVHostHandler(vhosts, handler)
+	return newGzipHandler(handler)
+}
+
 func newCorsHandler(srv http.Handler, allowedOrigins []string) http.Handler {
 	// disable CORS support if user has not specified a custom CORS configuration
 	if len(allowedOrigins) == 0 {
@@ -359,10 +367,11 @@ func newVHostHandler(vhosts []string, next http.Handler) http.Handler {
 	return &virtualHostHandler{vhostMap, next}
 }
 
-func newWebsocketUpgradeHandler(h http.Handler, ws http.Handler) http.Handler {
+func NewWebsocketUpgradeHandler(h http.Handler, ws http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if isWebsocket(r) {
 			ws.ServeHTTP(w, r)
+			log.Debug("serving websocket request")
 			return
 		}