p2p/simulations: use github.com/gorilla/websocket

This package was the last remaining user of golang.org/x/net/websocket.
Migrating to the new library wasn't straightforward because it is no
longer possible to treat WebSocket connections as a net.Conn.
This commit is contained in:
Felix Lange 2019-11-14 13:06:58 +01:00
parent e60a60ff21
commit 19081754d1
4 changed files with 45 additions and 32 deletions

View file

@ -41,7 +41,7 @@ import (
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"golang.org/x/net/websocket" "github.com/gorilla/websocket"
) )
func init() { func init() {
@ -118,7 +118,7 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
conf.Stack.P2P.NAT = nil conf.Stack.P2P.NAT = nil
conf.Stack.NoUSB = true 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) // initialise NodeConfig (usually a random port)
conf.Stack.P2P.ListenAddr = fmt.Sprintf(":%d", config.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 n.Cmd = cmd
// read the WebSocket address from the stderr logs // Wait for the node to start.
status := <-statusC status := <-statusC
if status.Err != "" { if status.Err != "" {
return errors.New(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 { if err != nil {
return fmt.Errorf("can't connect to RPC server: %v", err) return fmt.Errorf("can't connect to RPC server: %v", err)
} }
// node ready :) // Node ready :)
n.client = client n.client = client
n.wsAddr = status.WSEndpoint n.wsAddr = status.WSEndpoint
n.Info = status.NodeInfo n.Info = status.NodeInfo
@ -314,31 +314,37 @@ func (n *ExecNode) NodeInfo() *p2p.NodeInfo {
// ServeRPC serves RPC requests over the given connection by dialling the // ServeRPC serves RPC requests over the given connection by dialling the
// node's WebSocket address and joining the two connections // node's WebSocket address and joining the two connections
func (n *ExecNode) ServeRPC(clientConn net.Conn) error { func (n *ExecNode) ServeRPC(clientConn *websocket.Conn) error {
conn, err := websocket.Dial(n.wsAddr, "", "http://localhost") conn, _, err := websocket.DefaultDialer.Dial(n.wsAddr, nil)
if err != nil { if err != nil {
return err return err
} }
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(2) wg.Add(2)
join := func(src, dst net.Conn) { go wsCopy(&wg, conn, clientConn)
defer wg.Done() go wsCopy(&wg, clientConn, conn)
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 join(conn, clientConn)
go join(clientConn, conn)
wg.Wait() wg.Wait()
conn.Close()
return nil 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
}
}
}
// Snapshots creates snapshots of the services by calling the // Snapshots creates snapshots of the services by calling the
// simulation_snapshot RPC method // simulation_snapshot RPC method
func (n *ExecNode) Snapshots() (map[string][]byte, error) { func (n *ExecNode) Snapshots() (map[string][]byte, error) {

View file

@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/pipes" "github.com/ethereum/go-ethereum/p2p/simulations/pipes"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/gorilla/websocket"
) )
// SimAdapter is a NodeAdapter which creates in-memory simulation nodes and // 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 // ServeRPC serves RPC requests over the given connection by creating an
// in-memory client to the node's RPC server // in-memory client to the node's RPC server.
func (sn *SimNode) ServeRPC(conn net.Conn) error { func (sn *SimNode) ServeRPC(conn *websocket.Conn) error {
handler, err := sn.node.RPCHandler() handler, err := sn.node.RPCHandler()
if err != nil { if err != nil {
return err 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 return nil
} }

View file

@ -33,6 +33,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/p2p/enr"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/gorilla/websocket"
) )
// Node represents a node in a simulation network which is created by a // Node represents a node in a simulation network which is created by a
@ -51,7 +52,7 @@ type Node interface {
Client() (*rpc.Client, error) Client() (*rpc.Client, error)
// ServeRPC serves RPC requests over the given connection // 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 starts the node with the given snapshots
Start(snapshots map[string][]byte) error Start(snapshots map[string][]byte) error

View file

@ -34,8 +34,8 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/gorilla/websocket"
"github.com/julienschmidt/httprouter" "github.com/julienschmidt/httprouter"
"golang.org/x/net/websocket"
) )
// DefaultClient is the default simulation API client which expects the API // 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) 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 // NodeRPC forwards RPC requests to a node in the network via a WebSocket
// connection // connection
func (s *Server) NodeRPC(w http.ResponseWriter, req *http.Request) { func (s *Server) NodeRPC(w http.ResponseWriter, req *http.Request) {
node := req.Context().Value("node").(*Node) conn, err := wsUpgrade.Upgrade(w, req, nil)
if err != nil {
handler := func(conn *websocket.Conn) { return
node.ServeRPC(conn)
} }
defer conn.Close()
websocket.Server{Handler: handler}.ServeHTTP(w, req) node := req.Context().Value("node").(*Node)
node.ServeRPC(conn)
} }
// ServeHTTP implements the http.Handler interface by delegating to the // ServeHTTP implements the http.Handler interface by delegating to the