From e60a60ff21a6ac979de2b9aa6e9447863493cb4b Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 14 Nov 2019 13:04:51 +0100 Subject: [PATCH 1/4] rpc: improve codec abstraction rpc.ServerCodec is an opaque interface. There was only one way to get a codec using existing APIs: rpc.NewJSONCodec. This change exports newCodec (as NewFuncCodec) and NewJSONCodec (as NewCodec). It also makes all codec methods non-public to avoid showing internals in godoc. While here, remove codec options in tests because they are not supported anymore. --- rpc/client.go | 12 ++++++------ rpc/handler.go | 10 +++++----- rpc/http.go | 26 +++++++++++++------------- rpc/inproc.go | 4 ++-- rpc/ipc.go | 5 +++-- rpc/json.go | 40 ++++++++++++++++++++-------------------- rpc/server.go | 12 ++++++------ rpc/server_test.go | 2 +- rpc/stdio.go | 2 +- rpc/subscription.go | 4 ++-- rpc/subscription_test.go | 4 ++-- rpc/types.go | 10 +++++----- rpc/websocket.go | 4 ++-- 13 files changed, 68 insertions(+), 67 deletions(-) diff --git a/rpc/client.go b/rpc/client.go index 4b65d0042186a..a04198ad87832 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -117,7 +117,7 @@ func (c *Client) newClientConn(conn ServerCodec) *clientConn { func (cc *clientConn) close(err error, inflightReq *requestOp) { cc.handler.close(err, inflightReq) - cc.codec.Close() + cc.codec.close() } type readOp struct { @@ -484,7 +484,7 @@ func (c *Client) write(ctx context.Context, msg interface{}) error { return err } } - err := c.writeConn.Write(ctx, msg) + err := c.writeConn.writeJSON(ctx, msg) if err != nil { c.writeConn = nil } @@ -511,7 +511,7 @@ func (c *Client) reconnect(ctx context.Context) error { c.writeConn = newconn return nil case <-c.didClose: - newconn.Close() + newconn.close() return ErrClientQuit } } @@ -558,7 +558,7 @@ func (c *Client) dispatch(codec ServerCodec) { // Reconnect: case newcodec := <-c.reconnected: - log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.RemoteAddr()) + log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.remoteAddr()) if reading { // Wait for the previous read loop to exit. This is a rare case which // happens if this loop isn't notified in time after the connection breaks. @@ -612,9 +612,9 @@ func (c *Client) drainRead() { // read decodes RPC messages from a codec, feeding them into dispatch. func (c *Client) read(codec ServerCodec) { for { - msgs, batch, err := codec.Read() + msgs, batch, err := codec.readBatch() if _, ok := err.(*json.SyntaxError); ok { - codec.Write(context.Background(), errorMessage(&parseError{err.Error()})) + codec.writeJSON(context.Background(), errorMessage(&parseError{err.Error()})) } if err != nil { c.readErr <- err diff --git a/rpc/handler.go b/rpc/handler.go index 8a4e480ee2c71..461a88060982c 100644 --- a/rpc/handler.go +++ b/rpc/handler.go @@ -85,8 +85,8 @@ func newHandler(connCtx context.Context, conn jsonWriter, idgen func() ID, reg * serverSubs: make(map[ID]*Subscription), log: log.Root(), } - if conn.RemoteAddr() != "" { - h.log = h.log.New("conn", conn.RemoteAddr()) + if conn.remoteAddr() != "" { + h.log = h.log.New("conn", conn.remoteAddr()) } h.unsubscribeCb = newCallback(reflect.Value{}, reflect.ValueOf(h.unsubscribe)) return h @@ -97,7 +97,7 @@ func (h *handler) handleBatch(msgs []*jsonrpcMessage) { // Emit error response for empty batches: if len(msgs) == 0 { h.startCallProc(func(cp *callProc) { - h.conn.Write(cp.ctx, errorMessage(&invalidRequestError{"empty batch"})) + h.conn.writeJSON(cp.ctx, errorMessage(&invalidRequestError{"empty batch"})) }) return } @@ -122,7 +122,7 @@ func (h *handler) handleBatch(msgs []*jsonrpcMessage) { } h.addSubscriptions(cp.notifiers) if len(answers) > 0 { - h.conn.Write(cp.ctx, answers) + h.conn.writeJSON(cp.ctx, answers) } for _, n := range cp.notifiers { n.activate() @@ -139,7 +139,7 @@ func (h *handler) handleMsg(msg *jsonrpcMessage) { answer := h.handleCallMsg(cp, msg) h.addSubscriptions(cp.notifiers) if answer != nil { - h.conn.Write(cp.ctx, answer) + h.conn.writeJSON(cp.ctx, answer) } for _, n := range cp.notifiers { n.activate() diff --git a/rpc/http.go b/rpc/http.go index 2c0cb5edbe603..3b708603fec3e 100644 --- a/rpc/http.go +++ b/rpc/http.go @@ -47,29 +47,29 @@ type httpConn struct { client *http.Client req *http.Request closeOnce sync.Once - closed chan interface{} + closeCh chan interface{} } // httpConn is treated specially by Client. -func (hc *httpConn) Write(context.Context, interface{}) error { - panic("Write called on httpConn") +func (hc *httpConn) writeJSON(context.Context, interface{}) error { + panic("writeJSON called on httpConn") } -func (hc *httpConn) RemoteAddr() string { +func (hc *httpConn) remoteAddr() string { return hc.req.URL.String() } -func (hc *httpConn) Read() ([]*jsonrpcMessage, bool, error) { - <-hc.closed +func (hc *httpConn) readBatch() ([]*jsonrpcMessage, bool, error) { + <-hc.closeCh return nil, false, io.EOF } -func (hc *httpConn) Close() { - hc.closeOnce.Do(func() { close(hc.closed) }) +func (hc *httpConn) close() { + hc.closeOnce.Do(func() { close(hc.closeCh) }) } -func (hc *httpConn) Closed() <-chan interface{} { - return hc.closed +func (hc *httpConn) closed() <-chan interface{} { + return hc.closeCh } // HTTPTimeouts represents the configuration params for the HTTP RPC server. @@ -116,7 +116,7 @@ func DialHTTPWithClient(endpoint string, client *http.Client) (*Client, error) { initctx := context.Background() return newClient(initctx, func(context.Context) (ServerCodec, error) { - return &httpConn{client: client, req: req, closed: make(chan interface{})}, nil + return &httpConn{client: client, req: req, closeCh: make(chan interface{})}, nil }) } @@ -195,7 +195,7 @@ type httpServerConn struct { func newHTTPServerConn(r *http.Request, w http.ResponseWriter) ServerCodec { body := io.LimitReader(r.Body, maxRequestContentLength) conn := &httpServerConn{Reader: body, Writer: w, r: r} - return NewJSONCodec(conn) + return NewCodec(conn) } // Close does nothing and always returns nil. @@ -266,7 +266,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.Header().Set("content-type", contentType) codec := newHTTPServerConn(r, w) - defer codec.Close() + defer codec.close() s.serveSingleRequest(ctx, codec) } diff --git a/rpc/inproc.go b/rpc/inproc.go index c4456cfc4bfcf..fbe9a40ceca9f 100644 --- a/rpc/inproc.go +++ b/rpc/inproc.go @@ -26,8 +26,8 @@ func DialInProc(handler *Server) *Client { initctx := context.Background() c, _ := newClient(initctx, func(context.Context) (ServerCodec, error) { p1, p2 := net.Pipe() - go handler.ServeCodec(NewJSONCodec(p1), OptionMethodInvocation|OptionSubscriptions) - return NewJSONCodec(p2), nil + go handler.ServeCodec(NewCodec(p1), 0) + return NewCodec(p2), nil }) return c } diff --git a/rpc/ipc.go b/rpc/ipc.go index ad8ce03098f7b..b26d0058056c4 100644 --- a/rpc/ipc.go +++ b/rpc/ipc.go @@ -35,7 +35,7 @@ func (s *Server) ServeListener(l net.Listener) error { return err } log.Trace("Accepted RPC connection", "conn", conn.RemoteAddr()) - go s.ServeCodec(NewJSONCodec(conn), OptionMethodInvocation|OptionSubscriptions) + go s.ServeCodec(NewCodec(conn), 0) } } @@ -51,6 +51,7 @@ func DialIPC(ctx context.Context, endpoint string) (*Client, error) { if err != nil { return nil, err } - return NewJSONCodec(conn), err + return NewCodec(conn), err }) } + diff --git a/rpc/json.go b/rpc/json.go index 75c221038e61d..db2c1f7e3055d 100644 --- a/rpc/json.go +++ b/rpc/json.go @@ -164,43 +164,45 @@ func (c connWithRemoteAddr) RemoteAddr() string { return c.addr } // jsonCodec reads and writes JSON-RPC messages to the underlying connection. It also has // support for parsing arguments and serializing (result) objects. type jsonCodec struct { - remoteAddr string + remote string closer sync.Once // close closed channel once - closed chan interface{} // closed on Close + closeCh chan interface{} // closed on Close decode func(v interface{}) error // decoder to allow multiple transports encMu sync.Mutex // guards the encoder encode func(v interface{}) error // encoder to allow multiple transports conn deadlineCloser } -func newCodec(conn deadlineCloser, encode, decode func(v interface{}) error) ServerCodec { +// NewCodec creates a codec which uses the given functions to read and write. If conn +// implements ConnRemoteAddr, log messages will use it to include the remote address of +// the connection. +func NewFuncCodec(conn deadlineCloser, encode, decode func(v interface{}) error) ServerCodec { codec := &jsonCodec{ - closed: make(chan interface{}), + closeCh: make(chan interface{}), encode: encode, decode: decode, conn: conn, } if ra, ok := conn.(ConnRemoteAddr); ok { - codec.remoteAddr = ra.RemoteAddr() + codec.remote = ra.RemoteAddr() } return codec } -// NewJSONCodec creates a codec that reads from the given connection. If conn implements -// ConnRemoteAddr, log messages will use it to include the remote address of the -// connection. -func NewJSONCodec(conn Conn) ServerCodec { +// NewCodec creates a codec on the given connection. If conn implements ConnRemoteAddr, log +// messages will use it to include the remote address of the connection. +func NewCodec(conn Conn) ServerCodec { enc := json.NewEncoder(conn) dec := json.NewDecoder(conn) dec.UseNumber() - return newCodec(conn, enc.Encode, dec.Decode) + return NewFuncCodec(conn, enc.Encode, dec.Decode) } -func (c *jsonCodec) RemoteAddr() string { - return c.remoteAddr +func (c *jsonCodec) remoteAddr() string { + return c.remote } -func (c *jsonCodec) Read() (msg []*jsonrpcMessage, batch bool, err error) { +func (c *jsonCodec) readBatch() (msg []*jsonrpcMessage, batch bool, err error) { // Decode the next JSON object in the input stream. // This verifies basic syntax, etc. var rawmsg json.RawMessage @@ -211,8 +213,7 @@ func (c *jsonCodec) Read() (msg []*jsonrpcMessage, batch bool, err error) { return msg, batch, nil } -// Write sends a message to client. -func (c *jsonCodec) Write(ctx context.Context, v interface{}) error { +func (c *jsonCodec) writeJSON(ctx context.Context, v interface{}) error { c.encMu.Lock() defer c.encMu.Unlock() @@ -224,17 +225,16 @@ func (c *jsonCodec) Write(ctx context.Context, v interface{}) error { return c.encode(v) } -// Close the underlying connection -func (c *jsonCodec) Close() { +func (c *jsonCodec) close() { c.closer.Do(func() { - close(c.closed) + close(c.closeCh) c.conn.Close() }) } // Closed returns a channel which will be closed when Close is called -func (c *jsonCodec) Closed() <-chan interface{} { - return c.closed +func (c *jsonCodec) closed() <-chan interface{} { + return c.closeCh } // parseMessage parses raw bytes as a (batch of) JSON-RPC message(s). There are no error diff --git a/rpc/server.go b/rpc/server.go index 5a92847f20e4f..2e56112968857 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -66,13 +66,13 @@ func (s *Server) RegisterName(name string, receiver interface{}) error { return s.services.registerName(name, receiver) } -// ServeCodec reads incoming requests from codec, calls the appropriate callback and writes +// receiver reads incoming requests from codec, calls the appropriate callback and writes // the response back using the given codec. It will block until the codec is closed or the // server is stopped. In either case the codec is closed. // // Note that codec options are no longer supported. func (s *Server) ServeCodec(codec ServerCodec, options CodecOption) { - defer codec.Close() + defer codec.close() // Don't serve if server is stopped. if atomic.LoadInt32(&s.run) == 0 { @@ -84,7 +84,7 @@ func (s *Server) ServeCodec(codec ServerCodec, options CodecOption) { defer s.codecs.Remove(codec) c := initClient(codec, s.idgen, &s.services) - <-codec.Closed() + <-codec.closed() c.Close() } @@ -101,10 +101,10 @@ func (s *Server) serveSingleRequest(ctx context.Context, codec ServerCodec) { h.allowSubscribe = false defer h.close(io.EOF, nil) - reqs, batch, err := codec.Read() + reqs, batch, err := codec.readBatch() if err != nil { if err != io.EOF { - codec.Write(ctx, errorMessage(&invalidMessageError{"parse error"})) + codec.writeJSON(ctx, errorMessage(&invalidMessageError{"parse error"})) } return } @@ -122,7 +122,7 @@ func (s *Server) Stop() { if atomic.CompareAndSwapInt32(&s.run, 1, 0) { log.Debug("RPC server shutting down") s.codecs.Each(func(c interface{}) bool { - c.(ServerCodec).Close() + c.(ServerCodec).close() return true }) } diff --git a/rpc/server_test.go b/rpc/server_test.go index 39099546bbe0f..a4ca1fde43ea8 100644 --- a/rpc/server_test.go +++ b/rpc/server_test.go @@ -77,7 +77,7 @@ func runTestScript(t *testing.T, file string) { clientConn, serverConn := net.Pipe() defer clientConn.Close() - go server.ServeCodec(NewJSONCodec(serverConn), OptionMethodInvocation|OptionSubscriptions) + go server.ServeCodec(NewCodec(serverConn), 0) readbuf := bufio.NewReader(clientConn) for _, line := range strings.Split(string(content), "\n") { line = strings.TrimSpace(line) diff --git a/rpc/stdio.go b/rpc/stdio.go index d5dc066c99bb6..be2bab1c98bd1 100644 --- a/rpc/stdio.go +++ b/rpc/stdio.go @@ -33,7 +33,7 @@ func DialStdIO(ctx context.Context) (*Client, error) { // DialIO creates a client which uses the given IO channels func DialIO(ctx context.Context, in io.Reader, out io.Writer) (*Client, error) { return newClient(ctx, func(_ context.Context) (ServerCodec, error) { - return NewJSONCodec(stdioConn{ + return NewCodec(stdioConn{ in: in, out: out, }), nil diff --git a/rpc/subscription.go b/rpc/subscription.go index c1e869b8a3a46..153e24063e77f 100644 --- a/rpc/subscription.go +++ b/rpc/subscription.go @@ -141,7 +141,7 @@ func (n *Notifier) Notify(id ID, data interface{}) error { // Closed returns a channel that is closed when the RPC connection is closed. // Deprecated: use subscription error channel func (n *Notifier) Closed() <-chan interface{} { - return n.h.conn.Closed() + return n.h.conn.closed() } // takeSubscription returns the subscription (if one has been created). No subscription can @@ -172,7 +172,7 @@ func (n *Notifier) activate() error { func (n *Notifier) send(sub *Subscription, data json.RawMessage) error { params, _ := json.Marshal(&subscriptionResult{ID: string(sub.ID), Result: data}) ctx := context.Background() - return n.h.conn.Write(ctx, &jsonrpcMessage{ + return n.h.conn.writeJSON(ctx, &jsonrpcMessage{ Version: vsn, Method: n.namespace + notificationMethodSuffix, Params: params, diff --git a/rpc/subscription_test.go b/rpc/subscription_test.go index eba192450db09..c3a918a83295c 100644 --- a/rpc/subscription_test.go +++ b/rpc/subscription_test.go @@ -68,7 +68,7 @@ func TestSubscriptions(t *testing.T) { t.Fatalf("unable to register test service %v", err) } } - go server.ServeCodec(NewJSONCodec(serverConn), OptionMethodInvocation|OptionSubscriptions) + go server.ServeCodec(NewCodec(serverConn), 0) defer server.Stop() // wait for message and write them to the given channels @@ -130,7 +130,7 @@ func TestServerUnsubscribe(t *testing.T) { service := ¬ificationTestService{unsubscribed: make(chan string)} server.RegisterName("nftest2", service) p1, p2 := net.Pipe() - go server.ServeCodec(NewJSONCodec(p1), OptionMethodInvocation|OptionSubscriptions) + go server.ServeCodec(NewCodec(p1), 0) p2.SetDeadline(time.Now().Add(10 * time.Second)) diff --git a/rpc/types.go b/rpc/types.go index e6b9f2a300d22..fd783137ea116 100644 --- a/rpc/types.go +++ b/rpc/types.go @@ -45,19 +45,19 @@ type Error interface { // a RPC session. Implementations must be go-routine safe since the codec can be called in // multiple go-routines concurrently. type ServerCodec interface { - Read() (msgs []*jsonrpcMessage, isBatch bool, err error) - Close() + readBatch() (msgs []*jsonrpcMessage, isBatch bool, err error) + close() jsonWriter } // jsonWriter can write JSON messages to its underlying connection. // Implementations must be safe for concurrent use. type jsonWriter interface { - Write(context.Context, interface{}) error + writeJSON(context.Context, interface{}) error // Closed returns a channel which is closed when the connection is closed. - Closed() <-chan interface{} + closed() <-chan interface{} // RemoteAddr returns the peer address of the connection. - RemoteAddr() string + remoteAddr() string } type BlockNumber int64 diff --git a/rpc/websocket.go b/rpc/websocket.go index 1632d6af412a5..e34241e47fd05 100644 --- a/rpc/websocket.go +++ b/rpc/websocket.go @@ -63,7 +63,7 @@ func (s *Server) WebsocketHandler(allowedOrigins []string) http.Handler { return } codec := newWebsocketCodec(conn) - s.ServeCodec(codec, OptionMethodInvocation|OptionSubscriptions) + s.ServeCodec(codec, 0) }) } @@ -171,5 +171,5 @@ func wsClientHeaders(endpoint, origin string) (string, http.Header, error) { func newWebsocketCodec(conn *websocket.Conn) ServerCodec { conn.SetReadLimit(maxRequestContentLength) - return newCodec(conn, conn.WriteJSON, conn.ReadJSON) + return NewFuncCodec(conn, conn.WriteJSON, conn.ReadJSON) } From 19081754d125cd8f3c3432f7da7073d06d107c27 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 14 Nov 2019 13:06:58 +0100 Subject: [PATCH 2/4] 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. --- p2p/simulations/adapters/exec.go | 48 +++++++++++++++++------------- p2p/simulations/adapters/inproc.go | 8 +++-- p2p/simulations/adapters/types.go | 3 +- p2p/simulations/http.go | 18 ++++++----- 4 files changed, 45 insertions(+), 32 deletions(-) diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 5eb0a31566c62..0dc2ee21d7a97 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,31 +314,37 @@ 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 join(conn, clientConn) - go join(clientConn, conn) + 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 + } + } +} + // Snapshots creates snapshots of the services by calling the // simulation_snapshot RPC method func (n *ExecNode) Snapshots() (map[string][]byte, error) { diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index c1cf23a175c61..9787082e18259 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 850de96a155e7..ec22b712c351a 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 1f44cc66753aa..3f46baa7d3646 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 From 1c3e0688b05b3c19ac0da3a5a64b71cd329784cc Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 14 Nov 2019 13:59:54 +0100 Subject: [PATCH 3/4] vendor: delete golang.org/x/net/websocket --- vendor/golang.org/x/net/websocket/client.go | 106 ---- vendor/golang.org/x/net/websocket/dial.go | 24 - vendor/golang.org/x/net/websocket/hybi.go | 583 ------------------ vendor/golang.org/x/net/websocket/server.go | 113 ---- .../golang.org/x/net/websocket/websocket.go | 451 -------------- vendor/vendor.json | 6 - 6 files changed, 1283 deletions(-) delete mode 100644 vendor/golang.org/x/net/websocket/client.go delete mode 100644 vendor/golang.org/x/net/websocket/dial.go delete mode 100644 vendor/golang.org/x/net/websocket/hybi.go delete mode 100644 vendor/golang.org/x/net/websocket/server.go delete mode 100644 vendor/golang.org/x/net/websocket/websocket.go diff --git a/vendor/golang.org/x/net/websocket/client.go b/vendor/golang.org/x/net/websocket/client.go deleted file mode 100644 index 69a4ac7eefecc..0000000000000 --- a/vendor/golang.org/x/net/websocket/client.go +++ /dev/null @@ -1,106 +0,0 @@ -// Copyright 2009 The Go Authors. All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -package websocket - -import ( - "bufio" - "io" - "net" - "net/http" - "net/url" -) - -// DialError is an error that occurs while dialling a websocket server. -type DialError struct { - *Config - Err error -} - -func (e *DialError) Error() string { - return "websocket.Dial " + e.Config.Location.String() + ": " + e.Err.Error() -} - -// NewConfig creates a new WebSocket config for client connection. -func NewConfig(server, origin string) (config *Config, err error) { - config = new(Config) - config.Version = ProtocolVersionHybi13 - config.Location, err = url.ParseRequestURI(server) - if err != nil { - return - } - config.Origin, err = url.ParseRequestURI(origin) - if err != nil { - return - } - config.Header = http.Header(make(map[string][]string)) - return -} - -// NewClient creates a new WebSocket client connection over rwc. -func NewClient(config *Config, rwc io.ReadWriteCloser) (ws *Conn, err error) { - br := bufio.NewReader(rwc) - bw := bufio.NewWriter(rwc) - err = hybiClientHandshake(config, br, bw) - if err != nil { - return - } - buf := bufio.NewReadWriter(br, bw) - ws = newHybiClientConn(config, buf, rwc) - return -} - -// Dial opens a new client connection to a WebSocket. -func Dial(url_, protocol, origin string) (ws *Conn, err error) { - config, err := NewConfig(url_, origin) - if err != nil { - return nil, err - } - if protocol != "" { - config.Protocol = []string{protocol} - } - return DialConfig(config) -} - -var portMap = map[string]string{ - "ws": "80", - "wss": "443", -} - -func parseAuthority(location *url.URL) string { - if _, ok := portMap[location.Scheme]; ok { - if _, _, err := net.SplitHostPort(location.Host); err != nil { - return net.JoinHostPort(location.Host, portMap[location.Scheme]) - } - } - return location.Host -} - -// DialConfig opens a new client connection to a WebSocket with a config. -func DialConfig(config *Config) (ws *Conn, err error) { - var client net.Conn - if config.Location == nil { - return nil, &DialError{config, ErrBadWebSocketLocation} - } - if config.Origin == nil { - return nil, &DialError{config, ErrBadWebSocketOrigin} - } - dialer := config.Dialer - if dialer == nil { - dialer = &net.Dialer{} - } - client, err = dialWithDialer(dialer, config) - if err != nil { - goto Error - } - ws, err = NewClient(config, client) - if err != nil { - client.Close() - goto Error - } - return - -Error: - return nil, &DialError{config, err} -} diff --git a/vendor/golang.org/x/net/websocket/dial.go b/vendor/golang.org/x/net/websocket/dial.go deleted file mode 100644 index 2dab943a489a0..0000000000000 --- a/vendor/golang.org/x/net/websocket/dial.go +++ /dev/null @@ -1,24 +0,0 @@ -// Copyright 2015 The Go Authors. All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -package websocket - -import ( - "crypto/tls" - "net" -) - -func dialWithDialer(dialer *net.Dialer, config *Config) (conn net.Conn, err error) { - switch config.Location.Scheme { - case "ws": - conn, err = dialer.Dial("tcp", parseAuthority(config.Location)) - - case "wss": - conn, err = tls.DialWithDialer(dialer, "tcp", parseAuthority(config.Location), config.TlsConfig) - - default: - err = ErrBadScheme - } - return -} diff --git a/vendor/golang.org/x/net/websocket/hybi.go b/vendor/golang.org/x/net/websocket/hybi.go deleted file mode 100644 index 8cffdd16c91da..0000000000000 --- a/vendor/golang.org/x/net/websocket/hybi.go +++ /dev/null @@ -1,583 +0,0 @@ -// Copyright 2011 The Go Authors. All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -package websocket - -// This file implements a protocol of hybi draft. -// http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-17 - -import ( - "bufio" - "bytes" - "crypto/rand" - "crypto/sha1" - "encoding/base64" - "encoding/binary" - "fmt" - "io" - "io/ioutil" - "net/http" - "net/url" - "strings" -) - -const ( - websocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" - - closeStatusNormal = 1000 - closeStatusGoingAway = 1001 - closeStatusProtocolError = 1002 - closeStatusUnsupportedData = 1003 - closeStatusFrameTooLarge = 1004 - closeStatusNoStatusRcvd = 1005 - closeStatusAbnormalClosure = 1006 - closeStatusBadMessageData = 1007 - closeStatusPolicyViolation = 1008 - closeStatusTooBigData = 1009 - closeStatusExtensionMismatch = 1010 - - maxControlFramePayloadLength = 125 -) - -var ( - ErrBadMaskingKey = &ProtocolError{"bad masking key"} - ErrBadPongMessage = &ProtocolError{"bad pong message"} - ErrBadClosingStatus = &ProtocolError{"bad closing status"} - ErrUnsupportedExtensions = &ProtocolError{"unsupported extensions"} - ErrNotImplemented = &ProtocolError{"not implemented"} - - handshakeHeader = map[string]bool{ - "Host": true, - "Upgrade": true, - "Connection": true, - "Sec-Websocket-Key": true, - "Sec-Websocket-Origin": true, - "Sec-Websocket-Version": true, - "Sec-Websocket-Protocol": true, - "Sec-Websocket-Accept": true, - } -) - -// A hybiFrameHeader is a frame header as defined in hybi draft. -type hybiFrameHeader struct { - Fin bool - Rsv [3]bool - OpCode byte - Length int64 - MaskingKey []byte - - data *bytes.Buffer -} - -// A hybiFrameReader is a reader for hybi frame. -type hybiFrameReader struct { - reader io.Reader - - header hybiFrameHeader - pos int64 - length int -} - -func (frame *hybiFrameReader) Read(msg []byte) (n int, err error) { - n, err = frame.reader.Read(msg) - if frame.header.MaskingKey != nil { - for i := 0; i < n; i++ { - msg[i] = msg[i] ^ frame.header.MaskingKey[frame.pos%4] - frame.pos++ - } - } - return n, err -} - -func (frame *hybiFrameReader) PayloadType() byte { return frame.header.OpCode } - -func (frame *hybiFrameReader) HeaderReader() io.Reader { - if frame.header.data == nil { - return nil - } - if frame.header.data.Len() == 0 { - return nil - } - return frame.header.data -} - -func (frame *hybiFrameReader) TrailerReader() io.Reader { return nil } - -func (frame *hybiFrameReader) Len() (n int) { return frame.length } - -// A hybiFrameReaderFactory creates new frame reader based on its frame type. -type hybiFrameReaderFactory struct { - *bufio.Reader -} - -// NewFrameReader reads a frame header from the connection, and creates new reader for the frame. -// See Section 5.2 Base Framing protocol for detail. -// http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-17#section-5.2 -func (buf hybiFrameReaderFactory) NewFrameReader() (frame frameReader, err error) { - hybiFrame := new(hybiFrameReader) - frame = hybiFrame - var header []byte - var b byte - // First byte. FIN/RSV1/RSV2/RSV3/OpCode(4bits) - b, err = buf.ReadByte() - if err != nil { - return - } - header = append(header, b) - hybiFrame.header.Fin = ((header[0] >> 7) & 1) != 0 - for i := 0; i < 3; i++ { - j := uint(6 - i) - hybiFrame.header.Rsv[i] = ((header[0] >> j) & 1) != 0 - } - hybiFrame.header.OpCode = header[0] & 0x0f - - // Second byte. Mask/Payload len(7bits) - b, err = buf.ReadByte() - if err != nil { - return - } - header = append(header, b) - mask := (b & 0x80) != 0 - b &= 0x7f - lengthFields := 0 - switch { - case b <= 125: // Payload length 7bits. - hybiFrame.header.Length = int64(b) - case b == 126: // Payload length 7+16bits - lengthFields = 2 - case b == 127: // Payload length 7+64bits - lengthFields = 8 - } - for i := 0; i < lengthFields; i++ { - b, err = buf.ReadByte() - if err != nil { - return - } - if lengthFields == 8 && i == 0 { // MSB must be zero when 7+64 bits - b &= 0x7f - } - header = append(header, b) - hybiFrame.header.Length = hybiFrame.header.Length*256 + int64(b) - } - if mask { - // Masking key. 4 bytes. - for i := 0; i < 4; i++ { - b, err = buf.ReadByte() - if err != nil { - return - } - header = append(header, b) - hybiFrame.header.MaskingKey = append(hybiFrame.header.MaskingKey, b) - } - } - hybiFrame.reader = io.LimitReader(buf.Reader, hybiFrame.header.Length) - hybiFrame.header.data = bytes.NewBuffer(header) - hybiFrame.length = len(header) + int(hybiFrame.header.Length) - return -} - -// A HybiFrameWriter is a writer for hybi frame. -type hybiFrameWriter struct { - writer *bufio.Writer - - header *hybiFrameHeader -} - -func (frame *hybiFrameWriter) Write(msg []byte) (n int, err error) { - var header []byte - var b byte - if frame.header.Fin { - b |= 0x80 - } - for i := 0; i < 3; i++ { - if frame.header.Rsv[i] { - j := uint(6 - i) - b |= 1 << j - } - } - b |= frame.header.OpCode - header = append(header, b) - if frame.header.MaskingKey != nil { - b = 0x80 - } else { - b = 0 - } - lengthFields := 0 - length := len(msg) - switch { - case length <= 125: - b |= byte(length) - case length < 65536: - b |= 126 - lengthFields = 2 - default: - b |= 127 - lengthFields = 8 - } - header = append(header, b) - for i := 0; i < lengthFields; i++ { - j := uint((lengthFields - i - 1) * 8) - b = byte((length >> j) & 0xff) - header = append(header, b) - } - if frame.header.MaskingKey != nil { - if len(frame.header.MaskingKey) != 4 { - return 0, ErrBadMaskingKey - } - header = append(header, frame.header.MaskingKey...) - frame.writer.Write(header) - data := make([]byte, length) - for i := range data { - data[i] = msg[i] ^ frame.header.MaskingKey[i%4] - } - frame.writer.Write(data) - err = frame.writer.Flush() - return length, err - } - frame.writer.Write(header) - frame.writer.Write(msg) - err = frame.writer.Flush() - return length, err -} - -func (frame *hybiFrameWriter) Close() error { return nil } - -type hybiFrameWriterFactory struct { - *bufio.Writer - needMaskingKey bool -} - -func (buf hybiFrameWriterFactory) NewFrameWriter(payloadType byte) (frame frameWriter, err error) { - frameHeader := &hybiFrameHeader{Fin: true, OpCode: payloadType} - if buf.needMaskingKey { - frameHeader.MaskingKey, err = generateMaskingKey() - if err != nil { - return nil, err - } - } - return &hybiFrameWriter{writer: buf.Writer, header: frameHeader}, nil -} - -type hybiFrameHandler struct { - conn *Conn - payloadType byte -} - -func (handler *hybiFrameHandler) HandleFrame(frame frameReader) (frameReader, error) { - if handler.conn.IsServerConn() { - // The client MUST mask all frames sent to the server. - if frame.(*hybiFrameReader).header.MaskingKey == nil { - handler.WriteClose(closeStatusProtocolError) - return nil, io.EOF - } - } else { - // The server MUST NOT mask all frames. - if frame.(*hybiFrameReader).header.MaskingKey != nil { - handler.WriteClose(closeStatusProtocolError) - return nil, io.EOF - } - } - if header := frame.HeaderReader(); header != nil { - io.Copy(ioutil.Discard, header) - } - switch frame.PayloadType() { - case ContinuationFrame: - frame.(*hybiFrameReader).header.OpCode = handler.payloadType - case TextFrame, BinaryFrame: - handler.payloadType = frame.PayloadType() - case CloseFrame: - return nil, io.EOF - case PingFrame, PongFrame: - b := make([]byte, maxControlFramePayloadLength) - n, err := io.ReadFull(frame, b) - if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { - return nil, err - } - io.Copy(ioutil.Discard, frame) - if frame.PayloadType() == PingFrame { - if _, err := handler.WritePong(b[:n]); err != nil { - return nil, err - } - } - return nil, nil - } - return frame, nil -} - -func (handler *hybiFrameHandler) WriteClose(status int) (err error) { - handler.conn.wio.Lock() - defer handler.conn.wio.Unlock() - w, err := handler.conn.frameWriterFactory.NewFrameWriter(CloseFrame) - if err != nil { - return err - } - msg := make([]byte, 2) - binary.BigEndian.PutUint16(msg, uint16(status)) - _, err = w.Write(msg) - w.Close() - return err -} - -func (handler *hybiFrameHandler) WritePong(msg []byte) (n int, err error) { - handler.conn.wio.Lock() - defer handler.conn.wio.Unlock() - w, err := handler.conn.frameWriterFactory.NewFrameWriter(PongFrame) - if err != nil { - return 0, err - } - n, err = w.Write(msg) - w.Close() - return n, err -} - -// newHybiConn creates a new WebSocket connection speaking hybi draft protocol. -func newHybiConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn { - if buf == nil { - br := bufio.NewReader(rwc) - bw := bufio.NewWriter(rwc) - buf = bufio.NewReadWriter(br, bw) - } - ws := &Conn{config: config, request: request, buf: buf, rwc: rwc, - frameReaderFactory: hybiFrameReaderFactory{buf.Reader}, - frameWriterFactory: hybiFrameWriterFactory{ - buf.Writer, request == nil}, - PayloadType: TextFrame, - defaultCloseStatus: closeStatusNormal} - ws.frameHandler = &hybiFrameHandler{conn: ws} - return ws -} - -// generateMaskingKey generates a masking key for a frame. -func generateMaskingKey() (maskingKey []byte, err error) { - maskingKey = make([]byte, 4) - if _, err = io.ReadFull(rand.Reader, maskingKey); err != nil { - return - } - return -} - -// generateNonce generates a nonce consisting of a randomly selected 16-byte -// value that has been base64-encoded. -func generateNonce() (nonce []byte) { - key := make([]byte, 16) - if _, err := io.ReadFull(rand.Reader, key); err != nil { - panic(err) - } - nonce = make([]byte, 24) - base64.StdEncoding.Encode(nonce, key) - return -} - -// removeZone removes IPv6 zone identifer from host. -// E.g., "[fe80::1%en0]:8080" to "[fe80::1]:8080" -func removeZone(host string) string { - if !strings.HasPrefix(host, "[") { - return host - } - i := strings.LastIndex(host, "]") - if i < 0 { - return host - } - j := strings.LastIndex(host[:i], "%") - if j < 0 { - return host - } - return host[:j] + host[i:] -} - -// getNonceAccept computes the base64-encoded SHA-1 of the concatenation of -// the nonce ("Sec-WebSocket-Key" value) with the websocket GUID string. -func getNonceAccept(nonce []byte) (expected []byte, err error) { - h := sha1.New() - if _, err = h.Write(nonce); err != nil { - return - } - if _, err = h.Write([]byte(websocketGUID)); err != nil { - return - } - expected = make([]byte, 28) - base64.StdEncoding.Encode(expected, h.Sum(nil)) - return -} - -// Client handshake described in draft-ietf-hybi-thewebsocket-protocol-17 -func hybiClientHandshake(config *Config, br *bufio.Reader, bw *bufio.Writer) (err error) { - bw.WriteString("GET " + config.Location.RequestURI() + " HTTP/1.1\r\n") - - // According to RFC 6874, an HTTP client, proxy, or other - // intermediary must remove any IPv6 zone identifier attached - // to an outgoing URI. - bw.WriteString("Host: " + removeZone(config.Location.Host) + "\r\n") - bw.WriteString("Upgrade: websocket\r\n") - bw.WriteString("Connection: Upgrade\r\n") - nonce := generateNonce() - if config.handshakeData != nil { - nonce = []byte(config.handshakeData["key"]) - } - bw.WriteString("Sec-WebSocket-Key: " + string(nonce) + "\r\n") - bw.WriteString("Origin: " + strings.ToLower(config.Origin.String()) + "\r\n") - - if config.Version != ProtocolVersionHybi13 { - return ErrBadProtocolVersion - } - - bw.WriteString("Sec-WebSocket-Version: " + fmt.Sprintf("%d", config.Version) + "\r\n") - if len(config.Protocol) > 0 { - bw.WriteString("Sec-WebSocket-Protocol: " + strings.Join(config.Protocol, ", ") + "\r\n") - } - // TODO(ukai): send Sec-WebSocket-Extensions. - err = config.Header.WriteSubset(bw, handshakeHeader) - if err != nil { - return err - } - - bw.WriteString("\r\n") - if err = bw.Flush(); err != nil { - return err - } - - resp, err := http.ReadResponse(br, &http.Request{Method: "GET"}) - if err != nil { - return err - } - if resp.StatusCode != 101 { - return ErrBadStatus - } - if strings.ToLower(resp.Header.Get("Upgrade")) != "websocket" || - strings.ToLower(resp.Header.Get("Connection")) != "upgrade" { - return ErrBadUpgrade - } - expectedAccept, err := getNonceAccept(nonce) - if err != nil { - return err - } - if resp.Header.Get("Sec-WebSocket-Accept") != string(expectedAccept) { - return ErrChallengeResponse - } - if resp.Header.Get("Sec-WebSocket-Extensions") != "" { - return ErrUnsupportedExtensions - } - offeredProtocol := resp.Header.Get("Sec-WebSocket-Protocol") - if offeredProtocol != "" { - protocolMatched := false - for i := 0; i < len(config.Protocol); i++ { - if config.Protocol[i] == offeredProtocol { - protocolMatched = true - break - } - } - if !protocolMatched { - return ErrBadWebSocketProtocol - } - config.Protocol = []string{offeredProtocol} - } - - return nil -} - -// newHybiClientConn creates a client WebSocket connection after handshake. -func newHybiClientConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser) *Conn { - return newHybiConn(config, buf, rwc, nil) -} - -// A HybiServerHandshaker performs a server handshake using hybi draft protocol. -type hybiServerHandshaker struct { - *Config - accept []byte -} - -func (c *hybiServerHandshaker) ReadHandshake(buf *bufio.Reader, req *http.Request) (code int, err error) { - c.Version = ProtocolVersionHybi13 - if req.Method != "GET" { - return http.StatusMethodNotAllowed, ErrBadRequestMethod - } - // HTTP version can be safely ignored. - - if strings.ToLower(req.Header.Get("Upgrade")) != "websocket" || - !strings.Contains(strings.ToLower(req.Header.Get("Connection")), "upgrade") { - return http.StatusBadRequest, ErrNotWebSocket - } - - key := req.Header.Get("Sec-Websocket-Key") - if key == "" { - return http.StatusBadRequest, ErrChallengeResponse - } - version := req.Header.Get("Sec-Websocket-Version") - switch version { - case "13": - c.Version = ProtocolVersionHybi13 - default: - return http.StatusBadRequest, ErrBadWebSocketVersion - } - var scheme string - if req.TLS != nil { - scheme = "wss" - } else { - scheme = "ws" - } - c.Location, err = url.ParseRequestURI(scheme + "://" + req.Host + req.URL.RequestURI()) - if err != nil { - return http.StatusBadRequest, err - } - protocol := strings.TrimSpace(req.Header.Get("Sec-Websocket-Protocol")) - if protocol != "" { - protocols := strings.Split(protocol, ",") - for i := 0; i < len(protocols); i++ { - c.Protocol = append(c.Protocol, strings.TrimSpace(protocols[i])) - } - } - c.accept, err = getNonceAccept([]byte(key)) - if err != nil { - return http.StatusInternalServerError, err - } - return http.StatusSwitchingProtocols, nil -} - -// Origin parses the Origin header in req. -// If the Origin header is not set, it returns nil and nil. -func Origin(config *Config, req *http.Request) (*url.URL, error) { - var origin string - switch config.Version { - case ProtocolVersionHybi13: - origin = req.Header.Get("Origin") - } - if origin == "" { - return nil, nil - } - return url.ParseRequestURI(origin) -} - -func (c *hybiServerHandshaker) AcceptHandshake(buf *bufio.Writer) (err error) { - if len(c.Protocol) > 0 { - if len(c.Protocol) != 1 { - // You need choose a Protocol in Handshake func in Server. - return ErrBadWebSocketProtocol - } - } - buf.WriteString("HTTP/1.1 101 Switching Protocols\r\n") - buf.WriteString("Upgrade: websocket\r\n") - buf.WriteString("Connection: Upgrade\r\n") - buf.WriteString("Sec-WebSocket-Accept: " + string(c.accept) + "\r\n") - if len(c.Protocol) > 0 { - buf.WriteString("Sec-WebSocket-Protocol: " + c.Protocol[0] + "\r\n") - } - // TODO(ukai): send Sec-WebSocket-Extensions. - if c.Header != nil { - err := c.Header.WriteSubset(buf, handshakeHeader) - if err != nil { - return err - } - } - buf.WriteString("\r\n") - return buf.Flush() -} - -func (c *hybiServerHandshaker) NewServerConn(buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn { - return newHybiServerConn(c.Config, buf, rwc, request) -} - -// newHybiServerConn returns a new WebSocket connection speaking hybi draft protocol. -func newHybiServerConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn { - return newHybiConn(config, buf, rwc, request) -} diff --git a/vendor/golang.org/x/net/websocket/server.go b/vendor/golang.org/x/net/websocket/server.go deleted file mode 100644 index 0895dea1905aa..0000000000000 --- a/vendor/golang.org/x/net/websocket/server.go +++ /dev/null @@ -1,113 +0,0 @@ -// Copyright 2009 The Go Authors. All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -package websocket - -import ( - "bufio" - "fmt" - "io" - "net/http" -) - -func newServerConn(rwc io.ReadWriteCloser, buf *bufio.ReadWriter, req *http.Request, config *Config, handshake func(*Config, *http.Request) error) (conn *Conn, err error) { - var hs serverHandshaker = &hybiServerHandshaker{Config: config} - code, err := hs.ReadHandshake(buf.Reader, req) - if err == ErrBadWebSocketVersion { - fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code)) - fmt.Fprintf(buf, "Sec-WebSocket-Version: %s\r\n", SupportedProtocolVersion) - buf.WriteString("\r\n") - buf.WriteString(err.Error()) - buf.Flush() - return - } - if err != nil { - fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code)) - buf.WriteString("\r\n") - buf.WriteString(err.Error()) - buf.Flush() - return - } - if handshake != nil { - err = handshake(config, req) - if err != nil { - code = http.StatusForbidden - fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code)) - buf.WriteString("\r\n") - buf.Flush() - return - } - } - err = hs.AcceptHandshake(buf.Writer) - if err != nil { - code = http.StatusBadRequest - fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code)) - buf.WriteString("\r\n") - buf.Flush() - return - } - conn = hs.NewServerConn(buf, rwc, req) - return -} - -// Server represents a server of a WebSocket. -type Server struct { - // Config is a WebSocket configuration for new WebSocket connection. - Config - - // Handshake is an optional function in WebSocket handshake. - // For example, you can check, or don't check Origin header. - // Another example, you can select config.Protocol. - Handshake func(*Config, *http.Request) error - - // Handler handles a WebSocket connection. - Handler -} - -// ServeHTTP implements the http.Handler interface for a WebSocket -func (s Server) ServeHTTP(w http.ResponseWriter, req *http.Request) { - s.serveWebSocket(w, req) -} - -func (s Server) serveWebSocket(w http.ResponseWriter, req *http.Request) { - rwc, buf, err := w.(http.Hijacker).Hijack() - if err != nil { - panic("Hijack failed: " + err.Error()) - } - // The server should abort the WebSocket connection if it finds - // the client did not send a handshake that matches with protocol - // specification. - defer rwc.Close() - conn, err := newServerConn(rwc, buf, req, &s.Config, s.Handshake) - if err != nil { - return - } - if conn == nil { - panic("unexpected nil conn") - } - s.Handler(conn) -} - -// Handler is a simple interface to a WebSocket browser client. -// It checks if Origin header is valid URL by default. -// You might want to verify websocket.Conn.Config().Origin in the func. -// If you use Server instead of Handler, you could call websocket.Origin and -// check the origin in your Handshake func. So, if you want to accept -// non-browser clients, which do not send an Origin header, set a -// Server.Handshake that does not check the origin. -type Handler func(*Conn) - -func checkOrigin(config *Config, req *http.Request) (err error) { - config.Origin, err = Origin(config, req) - if err == nil && config.Origin == nil { - return fmt.Errorf("null origin") - } - return err -} - -// ServeHTTP implements the http.Handler interface for a WebSocket -func (h Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) { - s := Server{Handler: h, Handshake: checkOrigin} - s.serveWebSocket(w, req) -} diff --git a/vendor/golang.org/x/net/websocket/websocket.go b/vendor/golang.org/x/net/websocket/websocket.go deleted file mode 100644 index 1f4f7be40086d..0000000000000 --- a/vendor/golang.org/x/net/websocket/websocket.go +++ /dev/null @@ -1,451 +0,0 @@ -// Copyright 2009 The Go Authors. All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -// Package websocket implements a client and server for the WebSocket protocol -// as specified in RFC 6455. -// -// This package currently lacks some features found in an alternative -// and more actively maintained WebSocket package: -// -// https://godoc.org/github.com/gorilla/websocket -// -package websocket // import "golang.org/x/net/websocket" - -import ( - "bufio" - "crypto/tls" - "encoding/json" - "errors" - "io" - "io/ioutil" - "net" - "net/http" - "net/url" - "sync" - "time" -) - -const ( - ProtocolVersionHybi13 = 13 - ProtocolVersionHybi = ProtocolVersionHybi13 - SupportedProtocolVersion = "13" - - ContinuationFrame = 0 - TextFrame = 1 - BinaryFrame = 2 - CloseFrame = 8 - PingFrame = 9 - PongFrame = 10 - UnknownFrame = 255 - - DefaultMaxPayloadBytes = 32 << 20 // 32MB -) - -// ProtocolError represents WebSocket protocol errors. -type ProtocolError struct { - ErrorString string -} - -func (err *ProtocolError) Error() string { return err.ErrorString } - -var ( - ErrBadProtocolVersion = &ProtocolError{"bad protocol version"} - ErrBadScheme = &ProtocolError{"bad scheme"} - ErrBadStatus = &ProtocolError{"bad status"} - ErrBadUpgrade = &ProtocolError{"missing or bad upgrade"} - ErrBadWebSocketOrigin = &ProtocolError{"missing or bad WebSocket-Origin"} - ErrBadWebSocketLocation = &ProtocolError{"missing or bad WebSocket-Location"} - ErrBadWebSocketProtocol = &ProtocolError{"missing or bad WebSocket-Protocol"} - ErrBadWebSocketVersion = &ProtocolError{"missing or bad WebSocket Version"} - ErrChallengeResponse = &ProtocolError{"mismatch challenge/response"} - ErrBadFrame = &ProtocolError{"bad frame"} - ErrBadFrameBoundary = &ProtocolError{"not on frame boundary"} - ErrNotWebSocket = &ProtocolError{"not websocket protocol"} - ErrBadRequestMethod = &ProtocolError{"bad method"} - ErrNotSupported = &ProtocolError{"not supported"} -) - -// ErrFrameTooLarge is returned by Codec's Receive method if payload size -// exceeds limit set by Conn.MaxPayloadBytes -var ErrFrameTooLarge = errors.New("websocket: frame payload size exceeds limit") - -// Addr is an implementation of net.Addr for WebSocket. -type Addr struct { - *url.URL -} - -// Network returns the network type for a WebSocket, "websocket". -func (addr *Addr) Network() string { return "websocket" } - -// Config is a WebSocket configuration -type Config struct { - // A WebSocket server address. - Location *url.URL - - // A Websocket client origin. - Origin *url.URL - - // WebSocket subprotocols. - Protocol []string - - // WebSocket protocol version. - Version int - - // TLS config for secure WebSocket (wss). - TlsConfig *tls.Config - - // Additional header fields to be sent in WebSocket opening handshake. - Header http.Header - - // Dialer used when opening websocket connections. - Dialer *net.Dialer - - handshakeData map[string]string -} - -// serverHandshaker is an interface to handle WebSocket server side handshake. -type serverHandshaker interface { - // ReadHandshake reads handshake request message from client. - // Returns http response code and error if any. - ReadHandshake(buf *bufio.Reader, req *http.Request) (code int, err error) - - // AcceptHandshake accepts the client handshake request and sends - // handshake response back to client. - AcceptHandshake(buf *bufio.Writer) (err error) - - // NewServerConn creates a new WebSocket connection. - NewServerConn(buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) (conn *Conn) -} - -// frameReader is an interface to read a WebSocket frame. -type frameReader interface { - // Reader is to read payload of the frame. - io.Reader - - // PayloadType returns payload type. - PayloadType() byte - - // HeaderReader returns a reader to read header of the frame. - HeaderReader() io.Reader - - // TrailerReader returns a reader to read trailer of the frame. - // If it returns nil, there is no trailer in the frame. - TrailerReader() io.Reader - - // Len returns total length of the frame, including header and trailer. - Len() int -} - -// frameReaderFactory is an interface to creates new frame reader. -type frameReaderFactory interface { - NewFrameReader() (r frameReader, err error) -} - -// frameWriter is an interface to write a WebSocket frame. -type frameWriter interface { - // Writer is to write payload of the frame. - io.WriteCloser -} - -// frameWriterFactory is an interface to create new frame writer. -type frameWriterFactory interface { - NewFrameWriter(payloadType byte) (w frameWriter, err error) -} - -type frameHandler interface { - HandleFrame(frame frameReader) (r frameReader, err error) - WriteClose(status int) (err error) -} - -// Conn represents a WebSocket connection. -// -// Multiple goroutines may invoke methods on a Conn simultaneously. -type Conn struct { - config *Config - request *http.Request - - buf *bufio.ReadWriter - rwc io.ReadWriteCloser - - rio sync.Mutex - frameReaderFactory - frameReader - - wio sync.Mutex - frameWriterFactory - - frameHandler - PayloadType byte - defaultCloseStatus int - - // MaxPayloadBytes limits the size of frame payload received over Conn - // by Codec's Receive method. If zero, DefaultMaxPayloadBytes is used. - MaxPayloadBytes int -} - -// Read implements the io.Reader interface: -// it reads data of a frame from the WebSocket connection. -// if msg is not large enough for the frame data, it fills the msg and next Read -// will read the rest of the frame data. -// it reads Text frame or Binary frame. -func (ws *Conn) Read(msg []byte) (n int, err error) { - ws.rio.Lock() - defer ws.rio.Unlock() -again: - if ws.frameReader == nil { - frame, err := ws.frameReaderFactory.NewFrameReader() - if err != nil { - return 0, err - } - ws.frameReader, err = ws.frameHandler.HandleFrame(frame) - if err != nil { - return 0, err - } - if ws.frameReader == nil { - goto again - } - } - n, err = ws.frameReader.Read(msg) - if err == io.EOF { - if trailer := ws.frameReader.TrailerReader(); trailer != nil { - io.Copy(ioutil.Discard, trailer) - } - ws.frameReader = nil - goto again - } - return n, err -} - -// Write implements the io.Writer interface: -// it writes data as a frame to the WebSocket connection. -func (ws *Conn) Write(msg []byte) (n int, err error) { - ws.wio.Lock() - defer ws.wio.Unlock() - w, err := ws.frameWriterFactory.NewFrameWriter(ws.PayloadType) - if err != nil { - return 0, err - } - n, err = w.Write(msg) - w.Close() - return n, err -} - -// Close implements the io.Closer interface. -func (ws *Conn) Close() error { - err := ws.frameHandler.WriteClose(ws.defaultCloseStatus) - err1 := ws.rwc.Close() - if err != nil { - return err - } - return err1 -} - -// IsClientConn reports whether ws is a client-side connection. -func (ws *Conn) IsClientConn() bool { return ws.request == nil } - -// IsServerConn reports whether ws is a server-side connection. -func (ws *Conn) IsServerConn() bool { return ws.request != nil } - -// LocalAddr returns the WebSocket Origin for the connection for client, or -// the WebSocket location for server. -func (ws *Conn) LocalAddr() net.Addr { - if ws.IsClientConn() { - return &Addr{ws.config.Origin} - } - return &Addr{ws.config.Location} -} - -// RemoteAddr returns the WebSocket location for the connection for client, or -// the Websocket Origin for server. -func (ws *Conn) RemoteAddr() net.Addr { - if ws.IsClientConn() { - return &Addr{ws.config.Location} - } - return &Addr{ws.config.Origin} -} - -var errSetDeadline = errors.New("websocket: cannot set deadline: not using a net.Conn") - -// SetDeadline sets the connection's network read & write deadlines. -func (ws *Conn) SetDeadline(t time.Time) error { - if conn, ok := ws.rwc.(net.Conn); ok { - return conn.SetDeadline(t) - } - return errSetDeadline -} - -// SetReadDeadline sets the connection's network read deadline. -func (ws *Conn) SetReadDeadline(t time.Time) error { - if conn, ok := ws.rwc.(net.Conn); ok { - return conn.SetReadDeadline(t) - } - return errSetDeadline -} - -// SetWriteDeadline sets the connection's network write deadline. -func (ws *Conn) SetWriteDeadline(t time.Time) error { - if conn, ok := ws.rwc.(net.Conn); ok { - return conn.SetWriteDeadline(t) - } - return errSetDeadline -} - -// Config returns the WebSocket config. -func (ws *Conn) Config() *Config { return ws.config } - -// Request returns the http request upgraded to the WebSocket. -// It is nil for client side. -func (ws *Conn) Request() *http.Request { return ws.request } - -// Codec represents a symmetric pair of functions that implement a codec. -type Codec struct { - Marshal func(v interface{}) (data []byte, payloadType byte, err error) - Unmarshal func(data []byte, payloadType byte, v interface{}) (err error) -} - -// Send sends v marshaled by cd.Marshal as single frame to ws. -func (cd Codec) Send(ws *Conn, v interface{}) (err error) { - data, payloadType, err := cd.Marshal(v) - if err != nil { - return err - } - ws.wio.Lock() - defer ws.wio.Unlock() - w, err := ws.frameWriterFactory.NewFrameWriter(payloadType) - if err != nil { - return err - } - _, err = w.Write(data) - w.Close() - return err -} - -// Receive receives single frame from ws, unmarshaled by cd.Unmarshal and stores -// in v. The whole frame payload is read to an in-memory buffer; max size of -// payload is defined by ws.MaxPayloadBytes. If frame payload size exceeds -// limit, ErrFrameTooLarge is returned; in this case frame is not read off wire -// completely. The next call to Receive would read and discard leftover data of -// previous oversized frame before processing next frame. -func (cd Codec) Receive(ws *Conn, v interface{}) (err error) { - ws.rio.Lock() - defer ws.rio.Unlock() - if ws.frameReader != nil { - _, err = io.Copy(ioutil.Discard, ws.frameReader) - if err != nil { - return err - } - ws.frameReader = nil - } -again: - frame, err := ws.frameReaderFactory.NewFrameReader() - if err != nil { - return err - } - frame, err = ws.frameHandler.HandleFrame(frame) - if err != nil { - return err - } - if frame == nil { - goto again - } - maxPayloadBytes := ws.MaxPayloadBytes - if maxPayloadBytes == 0 { - maxPayloadBytes = DefaultMaxPayloadBytes - } - if hf, ok := frame.(*hybiFrameReader); ok && hf.header.Length > int64(maxPayloadBytes) { - // payload size exceeds limit, no need to call Unmarshal - // - // set frameReader to current oversized frame so that - // the next call to this function can drain leftover - // data before processing the next frame - ws.frameReader = frame - return ErrFrameTooLarge - } - payloadType := frame.PayloadType() - data, err := ioutil.ReadAll(frame) - if err != nil { - return err - } - return cd.Unmarshal(data, payloadType, v) -} - -func marshal(v interface{}) (msg []byte, payloadType byte, err error) { - switch data := v.(type) { - case string: - return []byte(data), TextFrame, nil - case []byte: - return data, BinaryFrame, nil - } - return nil, UnknownFrame, ErrNotSupported -} - -func unmarshal(msg []byte, payloadType byte, v interface{}) (err error) { - switch data := v.(type) { - case *string: - *data = string(msg) - return nil - case *[]byte: - *data = msg - return nil - } - return ErrNotSupported -} - -/* -Message is a codec to send/receive text/binary data in a frame on WebSocket connection. -To send/receive text frame, use string type. -To send/receive binary frame, use []byte type. - -Trivial usage: - - import "websocket" - - // receive text frame - var message string - websocket.Message.Receive(ws, &message) - - // send text frame - message = "hello" - websocket.Message.Send(ws, message) - - // receive binary frame - var data []byte - websocket.Message.Receive(ws, &data) - - // send binary frame - data = []byte{0, 1, 2} - websocket.Message.Send(ws, data) - -*/ -var Message = Codec{marshal, unmarshal} - -func jsonMarshal(v interface{}) (msg []byte, payloadType byte, err error) { - msg, err = json.Marshal(v) - return msg, TextFrame, err -} - -func jsonUnmarshal(msg []byte, payloadType byte, v interface{}) (err error) { - return json.Unmarshal(msg, v) -} - -/* -JSON is a codec to send/receive JSON data in a frame from a WebSocket connection. - -Trivial usage: - - import "websocket" - - type T struct { - Msg string - Count int - } - - // receive JSON type T - var data T - websocket.JSON.Receive(ws, &data) - - // send JSON type T - websocket.JSON.Send(ws, data) -*/ -var JSON = Codec{jsonMarshal, jsonUnmarshal} diff --git a/vendor/vendor.json b/vendor/vendor.json index 4acd5a7ff029e..f309e0da296b9 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -670,12 +670,6 @@ "revision": "da137c7871d730100384dbcf36e6f8fa493aef5b", "revisionTime": "2019-06-28T18:40:41Z" }, - { - "checksumSHA1": "F+tqxPGFt5x7DKZakbbMmENX1oQ=", - "path": "golang.org/x/net/websocket", - "revision": "da137c7871d730100384dbcf36e6f8fa493aef5b", - "revisionTime": "2019-06-28T18:40:41Z" - }, { "checksumSHA1": "4TEYFKrAUuwBMqExjQBsnf/CgjQ=", "path": "golang.org/x/sync/syncmap", From e903598c2277aa202aea6252ea0c0ccf5e5a8da5 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 15 Nov 2019 14:25:46 +0100 Subject: [PATCH 4/4] rpc: fix godoc comments and run gofmt --- rpc/http.go | 2 +- rpc/ipc.go | 1 - rpc/json.go | 22 +++++++++++----------- rpc/server.go | 2 +- 4 files changed, 13 insertions(+), 14 deletions(-) diff --git a/rpc/http.go b/rpc/http.go index 3b708603fec3e..0de127c80809d 100644 --- a/rpc/http.go +++ b/rpc/http.go @@ -47,7 +47,7 @@ type httpConn struct { client *http.Client req *http.Request closeOnce sync.Once - closeCh chan interface{} + closeCh chan interface{} } // httpConn is treated specially by Client. diff --git a/rpc/ipc.go b/rpc/ipc.go index b26d0058056c4..07a211c6277c4 100644 --- a/rpc/ipc.go +++ b/rpc/ipc.go @@ -54,4 +54,3 @@ func DialIPC(ctx context.Context, endpoint string) (*Client, error) { return NewCodec(conn), err }) } - diff --git a/rpc/json.go b/rpc/json.go index db2c1f7e3055d..ad7294d315607 100644 --- a/rpc/json.go +++ b/rpc/json.go @@ -164,24 +164,24 @@ func (c connWithRemoteAddr) RemoteAddr() string { return c.addr } // jsonCodec reads and writes JSON-RPC messages to the underlying connection. It also has // support for parsing arguments and serializing (result) objects. type jsonCodec struct { - remote string - closer sync.Once // close closed channel once - closeCh chan interface{} // closed on Close - decode func(v interface{}) error // decoder to allow multiple transports - encMu sync.Mutex // guards the encoder - encode func(v interface{}) error // encoder to allow multiple transports - conn deadlineCloser + remote string + closer sync.Once // close closed channel once + closeCh chan interface{} // closed on Close + decode func(v interface{}) error // decoder to allow multiple transports + encMu sync.Mutex // guards the encoder + encode func(v interface{}) error // encoder to allow multiple transports + conn deadlineCloser } -// NewCodec creates a codec which uses the given functions to read and write. If conn +// NewFuncCodec creates a codec which uses the given functions to read and write. If conn // implements ConnRemoteAddr, log messages will use it to include the remote address of // the connection. func NewFuncCodec(conn deadlineCloser, encode, decode func(v interface{}) error) ServerCodec { codec := &jsonCodec{ closeCh: make(chan interface{}), - encode: encode, - decode: decode, - conn: conn, + encode: encode, + decode: decode, + conn: conn, } if ra, ok := conn.(ConnRemoteAddr); ok { codec.remote = ra.RemoteAddr() diff --git a/rpc/server.go b/rpc/server.go index 2e56112968857..ddd7c744c0241 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -66,7 +66,7 @@ func (s *Server) RegisterName(name string, receiver interface{}) error { return s.services.registerName(name, receiver) } -// receiver reads incoming requests from codec, calls the appropriate callback and writes +// ServeCodec reads incoming requests from codec, calls the appropriate callback and writes // the response back using the given codec. It will block until the codec is closed or the // server is stopped. In either case the codec is closed. //