diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 5eb0a31566..0dc2ee21d7 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -41,7 +41,7 @@ import ( "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rpc" - "golang.org/x/net/websocket" + "github.com/gorilla/websocket" ) func init() { @@ -118,7 +118,7 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) { conf.Stack.P2P.NAT = nil conf.Stack.NoUSB = true - // listen on a localhost port, which we set when we + // Listen on a localhost port, which we set when we // initialise NodeConfig (usually a random port) conf.Stack.P2P.ListenAddr = fmt.Sprintf(":%d", config.Port) @@ -205,17 +205,17 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) { } n.Cmd = cmd - // read the WebSocket address from the stderr logs + // Wait for the node to start. status := <-statusC if status.Err != "" { return errors.New(status.Err) } - client, err := rpc.DialWebsocket(ctx, status.WSEndpoint, "http://localhost") + client, err := rpc.DialWebsocket(ctx, status.WSEndpoint, "") if err != nil { return fmt.Errorf("can't connect to RPC server: %v", err) } - // node ready :) + // Node ready :) n.client = client n.wsAddr = status.WSEndpoint n.Info = status.NodeInfo @@ -314,29 +314,35 @@ func (n *ExecNode) NodeInfo() *p2p.NodeInfo { // ServeRPC serves RPC requests over the given connection by dialling the // node's WebSocket address and joining the two connections -func (n *ExecNode) ServeRPC(clientConn net.Conn) error { - conn, err := websocket.Dial(n.wsAddr, "", "http://localhost") +func (n *ExecNode) ServeRPC(clientConn *websocket.Conn) error { + conn, _, err := websocket.DefaultDialer.Dial(n.wsAddr, nil) if err != nil { return err } var wg sync.WaitGroup wg.Add(2) - join := func(src, dst net.Conn) { - defer wg.Done() - io.Copy(dst, src) - // close the write end of the destination connection - if cw, ok := dst.(interface { - CloseWrite() error - }); ok { - cw.CloseWrite() - } else { - dst.Close() + go wsCopy(&wg, conn, clientConn) + go wsCopy(&wg, clientConn, conn) + wg.Wait() + conn.Close() + return nil +} + +func wsCopy(wg *sync.WaitGroup, src, dst *websocket.Conn) { + defer wg.Done() + for { + msgType, r, err := src.NextReader() + if err != nil { + return + } + w, err := dst.NextWriter(msgType) + if err != nil { + return + } + if _, err = io.Copy(w, r); err != nil { + return } } - go join(conn, clientConn) - go join(clientConn, conn) - wg.Wait() - return nil } // Snapshots creates snapshots of the services by calling the diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index c1cf23a175..9787082e18 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations/pipes" "github.com/ethereum/go-ethereum/rpc" + "github.com/gorilla/websocket" ) // SimAdapter is a NodeAdapter which creates in-memory simulation nodes and @@ -210,13 +211,14 @@ func (sn *SimNode) Client() (*rpc.Client, error) { } // ServeRPC serves RPC requests over the given connection by creating an -// in-memory client to the node's RPC server -func (sn *SimNode) ServeRPC(conn net.Conn) error { +// in-memory client to the node's RPC server. +func (sn *SimNode) ServeRPC(conn *websocket.Conn) error { handler, err := sn.node.RPCHandler() if err != nil { return err } - handler.ServeCodec(rpc.NewJSONCodec(conn), rpc.OptionMethodInvocation|rpc.OptionSubscriptions) + codec := rpc.NewFuncCodec(conn, conn.WriteJSON, conn.ReadJSON) + handler.ServeCodec(codec, 0) return nil } diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go index 850de96a15..ec22b712c3 100644 --- a/p2p/simulations/adapters/types.go +++ b/p2p/simulations/adapters/types.go @@ -33,6 +33,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/rpc" + "github.com/gorilla/websocket" ) // Node represents a node in a simulation network which is created by a @@ -51,7 +52,7 @@ type Node interface { Client() (*rpc.Client, error) // ServeRPC serves RPC requests over the given connection - ServeRPC(net.Conn) error + ServeRPC(*websocket.Conn) error // Start starts the node with the given snapshots Start(snapshots map[string][]byte) error diff --git a/p2p/simulations/http.go b/p2p/simulations/http.go index 1f44cc6675..3f46baa7d3 100644 --- a/p2p/simulations/http.go +++ b/p2p/simulations/http.go @@ -34,8 +34,8 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" + "github.com/gorilla/websocket" "github.com/julienschmidt/httprouter" - "golang.org/x/net/websocket" ) // DefaultClient is the default simulation API client which expects the API @@ -654,16 +654,20 @@ func (s *Server) Options(w http.ResponseWriter, req *http.Request) { w.WriteHeader(http.StatusOK) } +var wsUpgrade = websocket.Upgrader{ + CheckOrigin: func(*http.Request) bool { return true }, +} + // NodeRPC forwards RPC requests to a node in the network via a WebSocket // connection func (s *Server) NodeRPC(w http.ResponseWriter, req *http.Request) { - node := req.Context().Value("node").(*Node) - - handler := func(conn *websocket.Conn) { - node.ServeRPC(conn) + conn, err := wsUpgrade.Upgrade(w, req, nil) + if err != nil { + return } - - websocket.Server{Handler: handler}.ServeHTTP(w, req) + defer conn.Close() + node := req.Context().Value("node").(*Node) + node.ServeRPC(conn) } // ServeHTTP implements the http.Handler interface by delegating to the