diff --git a/internal/jsonrpc2/conn.go b/internal/jsonrpc2/conn.go index bc85191..0741696 100644 --- a/internal/jsonrpc2/conn.go +++ b/internal/jsonrpc2/conn.go @@ -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. diff --git a/internal/jsonrpc2/jsonrpc2_test.go b/internal/jsonrpc2/jsonrpc2_test.go index 64aff2e..5357982 100644 --- a/internal/jsonrpc2/jsonrpc2_test.go +++ b/internal/jsonrpc2/jsonrpc2_test.go @@ -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{} { diff --git a/internal/jsonrpc2/serve.go b/internal/jsonrpc2/serve.go index d24e243..94e90b4 100644 --- a/internal/jsonrpc2/serve.go +++ b/internal/jsonrpc2/serve.go @@ -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() } diff --git a/internal/jsonrpc2/serve_test.go b/internal/jsonrpc2/serve_test.go index 64985d7..5bb4aab 100644 --- a/internal/jsonrpc2/serve_test.go +++ b/internal/jsonrpc2/serve_test.go @@ -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()