func (c *Connection) Close() error {
	if err := c.closer.Close(); err != nil && !isClosingError(err) {
		return err
	}
	if err := c.Wait(); err != nil && !isClosingError(err) {
		return err
	}
	return nil
}
Close can be used to close the underlying stream, and then wait for the connection to
fully shut down.
This does not cancel in flight requests, but waits for them to gracefully complete.
package jsonrpc2
import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"sync/atomic"
	"github.com/govim/govim/cmd/govim/internal/golang_org_x_tools/event"
	"github.com/govim/govim/cmd/govim/internal/golang_org_x_tools/event/label"
	"github.com/govim/govim/cmd/govim/internal/golang_org_x_tools/lsp/debug/tag"
	errors "golang.org/x/xerrors"
)
type Binder interface {
	Bind(context.Context, *Connection) (ConnectionOptions, error)
}
type ConnectionOptions struct {
	Framer Framer
	Preempter Preempter
	Handler Handler
}
type Connection struct {
	seq         int64  
	closer      io.Closer
	writerBox   chan Writer
	outgoingBox chan map[ID]chan<- *Response
	incomingBox chan map[ID]*incoming
	async       *async
}
type AsyncCall struct {
	id        ID
	response  chan *Response  
	resultBox chan asyncResult
	endSpan   func()  
}
type asyncResult struct {
	result []byte
	err    error
}
type incoming struct {
	request   *Request         
	baseCtx   context.Context  
	done      func()           
	handleCtx context.Context  
	cancel    func()           
}
func (o ConnectionOptions) Bind(context.Context, *Connection) (ConnectionOptions, error) {
	return o, nil
}
func newConnection(ctx context.Context, rwc io.ReadWriteCloser, binder Binder) (*Connection, error) {
	c := &Connection{
		closer:      rwc,
		writerBox:   make(chan Writer, 1),
		outgoingBox: make(chan map[ID]chan<- *Response, 1),
		incomingBox: make(chan map[ID]*incoming, 1),
		async:       newAsync(),
	}
	options, err := binder.Bind(ctx, c)
	if err != nil {
		return nil, err
	}
	if options.Framer == nil {
		options.Framer = HeaderFramer()
	}
	if options.Preempter == nil {
		options.Preempter = defaultHandler{}
	}
	if options.Handler == nil {
		options.Handler = defaultHandler{}
	}
	c.outgoingBox <- make(map[ID]chan<- *Response)
	c.incomingBox <- make(map[ID]*incoming)
	reader := options.Framer.Reader(rwc)
	readToQueue := make(chan *incoming)
	queueToDeliver := make(chan *incoming)
	go c.readIncoming(ctx, reader, readToQueue)
	go c.manageQueue(ctx, options.Preempter, readToQueue, queueToDeliver)
	go c.deliverMessages(ctx, options.Handler, queueToDeliver)
	c.writerBox <- options.Framer.Writer(rwc)
	return c, nil
}
func (c *Connection) Notify(ctx context.Context, method string, params interface{}) error {
	notify, err := NewNotification(method, params)
	if err != nil {
		return errors.Errorf("marshaling notify parameters: %v", err)
	}
	ctx, done := event.Start(ctx, method,
		tag.Method.Of(method),
		tag.RPCDirection.Of(tag.Outbound),
	)
	event.Metric(ctx, tag.Started.Of(1))
	err = c.write(ctx, notify)
	switch {
	case err != nil:
		event.Label(ctx, tag.StatusCode.Of("ERROR"))
	default:
		event.Label(ctx, tag.StatusCode.Of("OK"))
	}
	done()
	return err
}
func (c *Connection) Call(ctx context.Context, method string, params interface{}) *AsyncCall {
	result := &AsyncCall{
		id:        Int64ID(atomic.AddInt64(&c.seq, 1)),
		resultBox: make(chan asyncResult, 1),
	}
	call, err := NewCall(result.id, method, params)
	if err != nil {
		result.resultBox <- asyncResult{err: errors.Errorf("marshaling call parameters: %w", err)}
		return result
	}
	ctx, endSpan := event.Start(ctx, method,
		tag.Method.Of(method),
		tag.RPCDirection.Of(tag.Outbound),
		tag.RPCID.Of(fmt.Sprintf("%q", result.id)),
	)
	result.endSpan = endSpan
	event.Metric(ctx, tag.Started.Of(1))
	result.response = make(chan *Response, 1)
	pending := <-c.outgoingBox
	pending[result.id] = result.response
	c.outgoingBox <- pending
	if err := c.write(ctx, call); err != nil {
		r, _ := NewResponse(result.id, nil, err)
		c.incomingResponse(r)
	}
	return result
}
func (a *AsyncCall) ID() ID { return a.id }
func (a *AsyncCall) IsReady() bool {
	select {
	case r := <-a.resultBox:
		a.resultBox <- r
		return true
	default:
		return false
	}
}
func (a *AsyncCall) Await(ctx context.Context, result interface{}) error {
	defer a.endSpan()
	var r asyncResult
	select {
	case response := <-a.response:
		switch {
		case response.Error != nil:
			r.err = response.Error
			event.Label(ctx, tag.StatusCode.Of("ERROR"))
		default:
			r.result = response.Result
			event.Label(ctx, tag.StatusCode.Of("OK"))
		}
	case r = <-a.resultBox:
	case <-ctx.Done():
		event.Label(ctx, tag.StatusCode.Of("CANCELLED"))
		return ctx.Err()
	}
	a.resultBox <- r
	if r.err != nil {
		return r.err
	}
	if result == nil || len(r.result) == 0 {
		return nil
	}
	return json.Unmarshal(r.result, result)
}
func (c *Connection) Respond(id ID, result interface{}, rerr error) error {
	pending := <-c.incomingBox
	defer func() { c.incomingBox <- pending }()
	entry, found := pending[id]
	if !found {
		return nil
	}
	delete(pending, id)
	return c.respond(entry, result, rerr)
}
func (c *Connection) Cancel(id ID) {
	pending := <-c.incomingBox
	defer func() { c.incomingBox <- pending }()
	if entry, found := pending[id]; found && entry.cancel != nil {
		entry.cancel()
		entry.cancel = nil
	}
}
func (c *Connection) Wait() error {
	return c.async.wait()
}