mirror of
https://github.com/sourcegraph/jsonrpc2.git
synced 2026-07-04 08:13:40 +02:00
Fix underlying connection not being closed on protocol error
Before this commit, the underlying connection of `Conn` was not being closed when a protocol error was encountered. This behavior contradicted with `Conn.DisconnectNotify()` because it reported that the underlying connection was being closed. Additionally, the underlying connection was now orphaned because `Conn` was no longer processing any of the subsequent requests. With this commit, the underlying connection is now being closed when a protocol error is encountered, matching what `Conn.DisconnectNotify()` has already been reporting.
This commit is contained in:
parent
78a3d790f3
commit
21db451b57
2 changed files with 49 additions and 33 deletions
59
jsonrpc2.go
59
jsonrpc2.go
|
|
@ -367,8 +367,7 @@ type Conn struct {
|
||||||
h Handler
|
h Handler
|
||||||
|
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
shutdown bool
|
closed bool
|
||||||
closing bool
|
|
||||||
seq uint64
|
seq uint64
|
||||||
pending map[ID]*call
|
pending map[ID]*call
|
||||||
|
|
||||||
|
|
@ -417,14 +416,35 @@ func NewConn(ctx context.Context, stream ObjectStream, h Handler, opts ...ConnOp
|
||||||
// Close closes the JSON-RPC connection. The connection may not be
|
// Close closes the JSON-RPC connection. The connection may not be
|
||||||
// used after it has been closed.
|
// used after it has been closed.
|
||||||
func (c *Conn) Close() error {
|
func (c *Conn) Close() error {
|
||||||
|
return c.close(nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Conn) close(cause error) error {
|
||||||
|
c.sending.Lock()
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
if c.shutdown || c.closing {
|
defer c.sending.Unlock()
|
||||||
c.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
|
|
||||||
|
if c.closed {
|
||||||
return ErrClosed
|
return ErrClosed
|
||||||
}
|
}
|
||||||
c.closing = true
|
|
||||||
c.mu.Unlock()
|
for _, call := range c.pending {
|
||||||
return c.stream.Close()
|
call.done <- cause
|
||||||
|
close(call.done)
|
||||||
|
}
|
||||||
|
|
||||||
|
if cause != nil && cause != io.EOF && cause != io.ErrUnexpectedEOF {
|
||||||
|
c.logger.Printf("jsonrpc2: protocol error: %v\n", cause)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := c.stream.Close(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
close(c.disconnect)
|
||||||
|
c.closed = true
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Conn) send(_ context.Context, m *anyMessage, wait bool) (cc *call, err error) {
|
func (c *Conn) send(_ context.Context, m *anyMessage, wait bool) (cc *call, err error) {
|
||||||
|
|
@ -436,7 +456,7 @@ func (c *Conn) send(_ context.Context, m *anyMessage, wait bool) (cc *call, err
|
||||||
var id ID
|
var id ID
|
||||||
|
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
if c.shutdown || c.closing {
|
if c.closed {
|
||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
return nil, ErrClosed
|
return nil, ErrClosed
|
||||||
}
|
}
|
||||||
|
|
@ -675,28 +695,7 @@ func (c *Conn) readMessages(ctx context.Context) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
c.close(err)
|
||||||
c.sending.Lock()
|
|
||||||
c.mu.Lock()
|
|
||||||
c.shutdown = true
|
|
||||||
closing := c.closing
|
|
||||||
if err == io.EOF {
|
|
||||||
if closing {
|
|
||||||
err = ErrClosed
|
|
||||||
} else {
|
|
||||||
err = io.ErrUnexpectedEOF
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, call := range c.pending {
|
|
||||||
call.done <- err
|
|
||||||
close(call.done)
|
|
||||||
}
|
|
||||||
c.mu.Unlock()
|
|
||||||
c.sending.Unlock()
|
|
||||||
if err != io.ErrUnexpectedEOF && !closing {
|
|
||||||
c.logger.Printf("jsonrpc2: protocol error: %v\n", err)
|
|
||||||
}
|
|
||||||
close(c.disconnect)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// call represents a JSON-RPC call over its entire lifecycle.
|
// call represents a JSON-RPC call over its entire lifecycle.
|
||||||
|
|
|
||||||
|
|
@ -390,6 +390,23 @@ func TestConn_Close_waitingForResponse(t *testing.T) {
|
||||||
<-done
|
<-done
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestConn_DisconnectNotify_protocol_error(t *testing.T) {
|
||||||
|
connA, connB := net.Pipe()
|
||||||
|
c := jsonrpc2.NewConn(context.Background(), jsonrpc2.NewBufferedStream(connB, jsonrpc2.VarintObjectCodec{}), nil)
|
||||||
|
connA.Write([]byte("invalid json"))
|
||||||
|
select {
|
||||||
|
case <-c.DisconnectNotify():
|
||||||
|
case <-time.After(200 * time.Millisecond):
|
||||||
|
t.Fatal("no disconnect notification")
|
||||||
|
}
|
||||||
|
// Assert that the underlying connection is closed by trying to write to it.
|
||||||
|
_, got := connB.Write(nil)
|
||||||
|
want := io.ErrClosedPipe
|
||||||
|
if got != want {
|
||||||
|
t.Fatalf("got %q, want %q", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func serve(ctx context.Context, lis net.Listener, h jsonrpc2.Handler, streamMaker streamMaker, opts ...jsonrpc2.ConnOpt) error {
|
func serve(ctx context.Context, lis net.Listener, h jsonrpc2.Handler, streamMaker streamMaker, opts ...jsonrpc2.ConnOpt) error {
|
||||||
for {
|
for {
|
||||||
conn, err := lis.Accept()
|
conn, err := lis.Accept()
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue