internal/jsonrpc2_v2: eliminate error return from Bind
Also make (*Connection).Close safe to call from Bind. A jsonrpc2_v2.Server has no good way to report an error from Bind. If the Server saves the error to return from its own Wait method, that might not ever actually happen: Wait waits for in-flight connections to complete, but if some existing connection stays up then Wait will not return. If the Server goes ahead with establishing the connection and installs its own Handler, that Handler needs to decide whether to serve the error from Bind or something more opaque, and at that point Bind may as well return a handler that makes that choice more precisely. If the Server merely logs the error and closes the Connection, then the Bind method itself may as well do that directly too. It seems to me that the only winning move is not to play. Only Bind is in a position to decide how to handle its errors appropriately, so it should not return them to the Server. Updates golang/go#56281. Change-Id: I07dc43ddf31253ce23da21a92d2b6c0f8d4b3afe Reviewed-on: https://go-review.googlesource.com/c/tools/+/443677 Run-TryBot: Bryan Mills <bcmills@google.com> Auto-Submit: Bryan Mills <bcmills@google.com> TryBot-Result: Gopher Robot <gobot@golang.org> Reviewed-by: Alan Donovan <adonovan@google.com> gopls-CI: kokoro <noreply+kokoro@google.com>
This commit is contained in:
parent
f882967b35
commit
f75270ffac
4 changed files with 24 additions and 30 deletions
|
|
@ -27,14 +27,16 @@ import (
|
|||
type Binder interface {
|
||||
// Bind returns the ConnectionOptions to use when establishing the passed-in
|
||||
// Connection.
|
||||
// The connection is not ready to use when Bind is called.
|
||||
Bind(context.Context, *Connection) (ConnectionOptions, error)
|
||||
//
|
||||
// The connection is not ready to use when Bind is called,
|
||||
// but Bind may close it without reading or writing to it.
|
||||
Bind(context.Context, *Connection) ConnectionOptions
|
||||
}
|
||||
|
||||
// A BinderFunc implements the Binder interface for a standalone Bind function.
|
||||
type BinderFunc func(context.Context, *Connection) (ConnectionOptions, error)
|
||||
type BinderFunc func(context.Context, *Connection) ConnectionOptions
|
||||
|
||||
func (f BinderFunc) Bind(ctx context.Context, c *Connection) (ConnectionOptions, error) {
|
||||
func (f BinderFunc) Bind(ctx context.Context, c *Connection) ConnectionOptions {
|
||||
return f(ctx, c)
|
||||
}
|
||||
|
||||
|
|
@ -130,8 +132,8 @@ type incomingRequest struct {
|
|||
}
|
||||
|
||||
// Bind returns the options unmodified.
|
||||
func (o ConnectionOptions) Bind(context.Context, *Connection) (ConnectionOptions, error) {
|
||||
return o, nil
|
||||
func (o ConnectionOptions) Bind(context.Context, *Connection) ConnectionOptions {
|
||||
return o
|
||||
}
|
||||
|
||||
// newConnection creates a new connection and runs it.
|
||||
|
|
@ -141,7 +143,7 @@ func (o ConnectionOptions) Bind(context.Context, *Connection) (ConnectionOptions
|
|||
// The connection is closed automatically (and its resources cleaned up) when
|
||||
// the last request has completed after the underlying ReadWriteCloser breaks,
|
||||
// but it may be stopped earlier by calling Close (for a clean shutdown).
|
||||
func newConnection(bindCtx context.Context, rwc io.ReadWriteCloser, binder Binder, onDone func()) (*Connection, error) {
|
||||
func newConnection(bindCtx context.Context, rwc io.ReadWriteCloser, binder Binder, onDone func()) *Connection {
|
||||
// TODO: Should we create a new event span here?
|
||||
// This will propagate cancellation from ctx; should it?
|
||||
ctx := notDone{bindCtx}
|
||||
|
|
@ -153,10 +155,7 @@ func newConnection(bindCtx context.Context, rwc io.ReadWriteCloser, binder Binde
|
|||
onDone: onDone,
|
||||
}
|
||||
|
||||
options, err := binder.Bind(bindCtx, c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
options := binder.Bind(bindCtx, c)
|
||||
framer := options.Framer
|
||||
if framer == nil {
|
||||
framer = HeaderFramer()
|
||||
|
|
@ -169,9 +168,14 @@ func newConnection(bindCtx context.Context, rwc io.ReadWriteCloser, binder Binde
|
|||
|
||||
c.writer <- framer.Writer(rwc)
|
||||
reader := framer.Reader(rwc)
|
||||
// The goroutines started here will continue until the underlying stream is closed.
|
||||
go c.readIncoming(ctx, reader, options.Preempter)
|
||||
return c, nil
|
||||
|
||||
c.updateInFlight(func(s *inFlightState) {
|
||||
if !s.closed {
|
||||
// The goroutine started here will continue until the underlying stream is closed.
|
||||
go c.readIncoming(ctx, reader, options.Preempter)
|
||||
}
|
||||
})
|
||||
return c
|
||||
}
|
||||
|
||||
// Notify invokes the target method but does not wait for a response.
|
||||
|
|
|
|||
|
|
@ -253,7 +253,7 @@ func verifyResults(t *testing.T, method string, results interface{}, expect inte
|
|||
}
|
||||
}
|
||||
|
||||
func (b binder) Bind(ctx context.Context, conn *jsonrpc2.Connection) (jsonrpc2.ConnectionOptions, error) {
|
||||
func (b binder) Bind(ctx context.Context, conn *jsonrpc2.Connection) jsonrpc2.ConnectionOptions {
|
||||
h := &handler{
|
||||
conn: conn,
|
||||
waiters: make(chan map[string]chan struct{}, 1),
|
||||
|
|
@ -267,7 +267,7 @@ func (b binder) Bind(ctx context.Context, conn *jsonrpc2.Connection) (jsonrpc2.C
|
|||
Framer: b.framer,
|
||||
Preempter: h,
|
||||
Handler: h,
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (h *handler) waiter(name string) chan struct{} {
|
||||
|
|
|
|||
|
|
@ -60,12 +60,7 @@ func Dial(ctx context.Context, dialer Dialer, binder Binder) (*Connection, error
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conn, err := newConnection(ctx, rwc, binder, nil)
|
||||
if err != nil {
|
||||
rwc.Close()
|
||||
return nil, err
|
||||
}
|
||||
return conn, nil
|
||||
return newConnection(ctx, rwc, binder, nil), nil
|
||||
}
|
||||
|
||||
// Serve starts a new server listening for incoming connections and returns
|
||||
|
|
@ -120,12 +115,7 @@ func (s *Server) run(ctx context.Context) {
|
|||
|
||||
// A new inbound connection.
|
||||
activeConns.Add(1)
|
||||
_, err = newConnection(ctx, rwc, s.binder, activeConns.Done) // unregisters itself when done
|
||||
if err != nil {
|
||||
rwc.Close()
|
||||
activeConns.Done()
|
||||
s.async.setError(err)
|
||||
}
|
||||
_ = newConnection(ctx, rwc, s.binder, activeConns.Done) // unregisters itself when done
|
||||
}
|
||||
activeConns.Wait()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -288,7 +288,7 @@ func TestCloseCallRace(t *testing.T) {
|
|||
|
||||
pokec := make(chan *jsonrpc2.AsyncCall, 1)
|
||||
|
||||
s, err := jsonrpc2.Serve(ctx, listener, jsonrpc2.BinderFunc(func(_ context.Context, srvConn *jsonrpc2.Connection) (jsonrpc2.ConnectionOptions, error) {
|
||||
s, err := jsonrpc2.Serve(ctx, listener, jsonrpc2.BinderFunc(func(_ context.Context, srvConn *jsonrpc2.Connection) jsonrpc2.ConnectionOptions {
|
||||
h := jsonrpc2.HandlerFunc(func(ctx context.Context, _ *jsonrpc2.Request) (interface{}, error) {
|
||||
// Start a concurrent call from the server to the client.
|
||||
// The point of this test is to ensure this doesn't deadlock
|
||||
|
|
@ -303,7 +303,7 @@ func TestCloseCallRace(t *testing.T) {
|
|||
|
||||
return &msg{"pong"}, nil
|
||||
})
|
||||
return jsonrpc2.ConnectionOptions{Handler: h}, nil
|
||||
return jsonrpc2.ConnectionOptions{Handler: h}
|
||||
}))
|
||||
if err != nil {
|
||||
listener.Close()
|
||||
|
|
|
|||
Loading…
Reference in a new issue