From e13d78db5252073dcbadb2198c12c9a8a16c1b00 Mon Sep 17 00:00:00 2001 From: renaynay <41963722+renaynay@users.noreply.github.com> Date: Tue, 24 Mar 2020 20:24:22 +0100 Subject: [PATCH] dirty functional implementation, some tests broken, some commented out --- cmd/geth/retesteth.go | 2 +- graphql/service.go | 20 ++++++++++---------- node/api.go | 2 +- node/node.go | 25 +++++++++++++++++-------- rpc/endpoints.go | 6 ++++-- rpc/http.go | 19 ++++++++++++++++++- rpc/websocket.go | 1 + 7 files changed, 52 insertions(+), 23 deletions(-) diff --git a/cmd/geth/retesteth.go b/cmd/geth/retesteth.go index 629c2ea7f2..c4013db422 100644 --- a/cmd/geth/retesteth.go +++ b/cmd/geth/retesteth.go @@ -895,7 +895,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) + listener, _, err := rpc.StartHTTPEndpoint(httpEndpoint, rpcAPI, []string{"test", "eth", "debug", "web3"}, cors, vhosts, RetestethHTTPTimeouts, []string{}) if err != nil { utils.Fatalf("Could not start RPC api: %v", err) } diff --git a/graphql/service.go b/graphql/service.go index f640756806..70d75ae72f 100644 --- a/graphql/service.go +++ b/graphql/service.go @@ -60,16 +60,16 @@ func (s *Service) APIs() []rpc.API { return nil } // Start is called after all services have been constructed and the networking // layer was also initialized to spawn any goroutines required by the service. func (s *Service) Start(server *p2p.Server) error { - var err error - s.handler, err = newHandler(s.backend) - if err != nil { - return err - } - if s.listener, err = net.Listen("tcp", s.endpoint); err != nil { - return err - } - go rpc.NewHTTPServer(s.cors, s.vhosts, s.timeouts, s.handler).Serve(s.listener) - log.Info("GraphQL endpoint opened", "url", fmt.Sprintf("http://%s", s.endpoint)) + //var err error + //s.handler, err = newHandler(s.backend) + //if err != nil { + // return err + //} + //if s.listener, err = net.Listen("tcp", s.endpoint); err != nil { + // return err + //} + //go rpc.NewHTTPServer(s.cors, s.vhosts, s.timeouts, s.handler, []string{}).Serve(s.listener) + //log.Info("GraphQL endpoint opened", "url", fmt.Sprintf("http://%s", s.endpoint)) return nil } diff --git a/node/api.go b/node/api.go index 66cd1dde33..1a73d1321d 100644 --- a/node/api.go +++ b/node/api.go @@ -186,7 +186,7 @@ func (api *PrivateAdminAPI) StartRPC(host *string, port *int, cors *string, apis } } - if err := api.node.startHTTP(fmt.Sprintf("%s:%d", *host, *port), api.node.rpcAPIs, modules, allowedOrigins, allowedVHosts, api.node.config.HTTPTimeouts); err != nil { + if err := api.node.startHTTP(fmt.Sprintf("%s:%d", *host, *port), api.node.rpcAPIs, modules, allowedOrigins, allowedVHosts, api.node.config.HTTPTimeouts, api.node.config.WSOrigins); err != nil { return false, err } return true, nil diff --git a/node/node.go b/node/node.go index 7d8f9b07b9..c3d1df78e5 100644 --- a/node/node.go +++ b/node/node.go @@ -291,17 +291,21 @@ func (n *Node) startRPC(services map[reflect.Type]Service) error { n.stopInProc() return err } - if err := n.startHTTP(n.httpEndpoint, apis, n.config.HTTPModules, n.config.HTTPCors, n.config.HTTPVirtualHosts, n.config.HTTPTimeouts); err != nil { + if err := n.startHTTP(n.httpEndpoint, apis, n.config.HTTPModules, n.config.HTTPCors, n.config.HTTPVirtualHosts, n.config.HTTPTimeouts, n.config.WSOrigins); err != nil { n.stopIPC() n.stopInProc() return err } - if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil { - n.stopHTTP() - n.stopIPC() - n.stopInProc() - return err + // if endpoints are not the same, start separate servers + if n.httpEndpoint != n.wsEndpoint { + if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil { + n.stopHTTP() + n.stopIPC() + n.stopInProc() + return err + } } + // All API endpoints started successfully n.rpcAPIs = apis return nil @@ -359,18 +363,23 @@ func (n *Node) stopIPC() { } // startHTTP initializes and starts the HTTP RPC endpoint. -func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors []string, vhosts []string, timeouts rpc.HTTPTimeouts) error { +func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors []string, vhosts []string, timeouts rpc.HTTPTimeouts, wsOrigins []string) error { // Short circuit if the HTTP endpoint isn't being exposed if endpoint == "" { return nil } - listener, handler, err := rpc.StartHTTPEndpoint(endpoint, apis, modules, cors, vhosts, timeouts) + listener, handler, err := rpc.StartHTTPEndpoint(endpoint, apis, modules, cors, vhosts, timeouts, wsOrigins) if err != nil { return err } n.log.Info("HTTP endpoint opened", "url", fmt.Sprintf("http://%v/", listener.Addr()), "cors", strings.Join(cors, ","), "vhosts", strings.Join(vhosts, ",")) + + if n.httpEndpoint == n.wsEndpoint { + n.log.Info("WebSocket endpoint opened", "url", fmt.Sprintf("ws://%v", listener.Addr())) + } + // All listeners booted successfully n.httpEndpoint = endpoint n.httpListener = listener diff --git a/rpc/endpoints.go b/rpc/endpoints.go index 09f389d71b..4a0ded76db 100644 --- a/rpc/endpoints.go +++ b/rpc/endpoints.go @@ -42,7 +42,7 @@ 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) (net.Listener, *Server, error) { +func StartHTTPEndpoint(endpoint string, apis []API, modules []string, cors []string, vhosts []string, timeouts HTTPTimeouts, wsOrigins []string) (net.Listener, *Server, error) { if bad, available := checkModuleAvailability(modules, apis); len(bad) > 0 { log.Error("Unavailable modules in HTTP API list", "unavailable", bad, "available", available) } @@ -69,7 +69,9 @@ func StartHTTPEndpoint(endpoint string, apis []API, modules []string, cors []str if listener, err = net.Listen("tcp", endpoint); err != nil { return nil, nil, err } - go NewHTTPServer(cors, vhosts, timeouts, handler).Serve(listener) + + + go NewHTTPServer(cors, vhosts, timeouts, handler, handler.WebsocketHandler(wsOrigins)).Serve(listener) return listener, handler, err } diff --git a/rpc/http.go b/rpc/http.go index 40810c7b44..439029d44b 100644 --- a/rpc/http.go +++ b/rpc/http.go @@ -212,11 +212,12 @@ func (t *httpServerConn) SetWriteDeadline(time.Time) error { return nil } // NewHTTPServer creates a new HTTP RPC server around an API provider. // // Deprecated: Server implements http.Handler -func NewHTTPServer(cors []string, vhosts []string, timeouts HTTPTimeouts, srv http.Handler) *http.Server { +func NewHTTPServer(cors []string, vhosts []string, timeouts HTTPTimeouts, srv http.Handler, ws http.Handler) *http.Server { // Wrap the CORS-handler within a host-handler handler := newCorsHandler(srv, cors) handler = newVHostHandler(vhosts, handler) handler = newGzipHandler(handler) + handler = newWebsocketUpgradeHandler(handler, ws) // Make sure timeout values are meaningful if timeouts.ReadTimeout < time.Second { @@ -357,3 +358,19 @@ func newVHostHandler(vhosts []string, next http.Handler) http.Handler { } return &virtualHostHandler{vhostMap, next} } + +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) + return + } + + h.ServeHTTP(w, r) + }) +} + +func isWebsocket(r *http.Request) bool { + return strings.ToLower(r.Header.Get("Upgrade")) == "websocket" && + strings.ToLower(r.Header.Get("Connection")) == "upgrade" +} diff --git a/rpc/websocket.go b/rpc/websocket.go index b7ec56c6a6..b49553c160 100644 --- a/rpc/websocket.go +++ b/rpc/websocket.go @@ -62,6 +62,7 @@ func (s *Server) WebsocketHandler(allowedOrigins []string) http.Handler { log.Debug("WebSocket upgrade failed", "err", err) return } + codec := newWebsocketCodec(conn) s.ServeCodec(codec, 0) })