all: sync to golang/tools@dcbf2a9ed15d

Author: zchee

conn.go

@@ -1,5 +1,5 @@
-// Copyright 2020 The Go Language Server Authors.
 // SPDX-License-Identifier: BSD-3-Clause
+// SPDX-FileCopyrightText: Copyright 2021 The Go Language Server Authors
 
 package jsonrpc2
 
@@ -8,10 +8,13 @@ import (
 	"fmt"
 	"sync"
 	"sync/atomic"
+
+	"go.lsp.dev/pkg/event"
+	"go.lsp.dev/pkg/event/label"
+	"go.lsp.dev/pkg/event/tag"
 )
 
 // Conn is the common interface to jsonrpc clients and servers.
-//
 // Conn is bidirectional; it does not have a designated server or client end.
 // It manages the jsonrpc2 protocol, connecting responses back to their calls.
 type Conn interface {
@@ -52,7 +55,7 @@ type Conn interface {
 }
 
 type conn struct {
-	seq       int64      // access atomic
+	seq       int64      // access atomically
 	writeMu   sync.Mutex // protects writes to the stream
 	stream    Stream
 	pendingMu sync.Mutex // protects the pending map
@@ -69,7 +72,6 @@ func NewConn(s Stream) Conn {
 		pending: make(map[ID]chan *Response),
 		done:    make(chan struct{}),
 	}
-
 	return conn
 }
 
@@ -79,41 +81,51 @@ func (c *conn) Notify(ctx context.Context, method string, params interface{}) (e
 	if err != nil {
 		return fmt.Errorf("marshaling notify parameters: %w", err)
 	}
-
-	_, err = c.write(ctx, notify)
+	ctx, done := event.Start(ctx, method,
+		tag.Method.Of(method),
+		tag.RPCDirection.Of(tag.Outbound),
+	)
+	defer func() {
+		recordStatus(ctx, err)
+		done()
+	}()
+
+	event.Metric(ctx, tag.Started.Of(1))
+	n, err := c.write(ctx, notify)
+	event.Metric(ctx, tag.SentBytes.Of(n))
 	return err
 }
 
-func (c *conn) replier(req Message) Replier {
-	reply := func(ctx context.Context, result interface{}, err error) error {
-		call, ok := req.(*Request)
+func (c *conn) replier(req Message, spanDone func()) Replier {
+	return func(ctx context.Context, result interface{}, err error) error {
+		defer func() {
+			recordStatus(ctx, err)
+			spanDone()
+		}()
+		call, ok := req.(*Call)
 		if !ok {
 			// request was a notify, no need to respond
 			return nil
 		}
-
 		response, err := NewResponse(call.id, result, err)
 		if err != nil {
 			return err
 		}
-
-		_, err = c.write(ctx, response)
+		n, err := c.write(ctx, response)
+		event.Metric(ctx, tag.SentBytes.Of(n))
 		if err != nil {
 			// TODO(iancottrell): if a stream write fails, we really need to shut down
 			// the whole stream
 			return err
 		}
 		return nil
 	}
-
-	return reply
 }
 
-func (c *conn) write(ctx context.Context, msg Message) (n int64, err error) {
+func (c *conn) write(ctx context.Context, msg Message) (int64, error) {
 	c.writeMu.Lock()
-	n, err = c.stream.Write(ctx, msg)
-	c.writeMu.Unlock()
-	return
+	defer c.writeMu.Unlock()
+	return c.stream.Write(ctx, msg)
 }
 
 // Go implemens Conn.
@@ -123,24 +135,34 @@ func (c *conn) Go(ctx context.Context, handler Handler) {
 
 func (c *conn) run(ctx context.Context, handler Handler) {
 	defer close(c.done)
-
 	for {
 		// get the next message
-		msg, _, err := c.stream.Read(ctx)
+		msg, n, err := c.stream.Read(ctx)
 		if err != nil {
 			// The stream failed, we cannot continue.
 			c.fail(err)
 			return
 		}
-
 		switch msg := msg.(type) {
-		case Requester:
-			if err := handler(ctx, c.replier(msg), msg); err != nil {
+		case Request:
+			labels := []label.Label{
+				tag.Method.Of(msg.Method()),
+				tag.RPCDirection.Of(tag.Inbound),
+				{}, // reserved for ID if present
+			}
+			if call, ok := msg.(*Call); ok {
+				labels[len(labels)-1] = tag.RPCID.Of(fmt.Sprintf("%q", call.ID()))
+			} else {
+				labels = labels[:len(labels)-1]
+			}
+			reqCtx, spanDone := event.Start(ctx, msg.Method(), labels...)
+			event.Metric(reqCtx,
+				tag.Started.Of(1),
+				tag.ReceivedBytes.Of(n))
+			if err := handler(reqCtx, c.replier(msg, spanDone), msg); err != nil {
 				// delivery failed, not much we can do
-				c.fail(err)
-				return
+				event.Error(reqCtx, "jsonrpc2 message delivery failed", err)
 			}
-
 		case *Response:
 			// If method is not set, this should be a response, in which case we must
 			// have an id to send the response back to the caller.
@@ -177,3 +199,11 @@ func (c *conn) fail(err error) {
 	c.err.Store(err)
 	c.stream.Close()
 }
+
+func recordStatus(ctx context.Context, err error) {
+	if err != nil {
+		event.Label(ctx, tag.StatusCode.Of("ERROR"))
+	} else {
+		event.Label(ctx, tag.StatusCode.Of("OK"))
+	}
+}

conn_gojay.go

@@ -18,7 +18,7 @@ import (
 func (c *conn) Call(ctx context.Context, method string, params, result interface{}) (ID, error) {
 	// generate a new request identifier
 	id := ID{number: atomic.AddInt64(&c.seq, 1)}
-	call, err := NewRequest(id, method, params)
+	call, err := NewCall(id, method, params)
 	if err != nil {
 		return id, fmt.Errorf("marshaling call parameters: %v", err)
 	}

conn_json.go

@@ -1,5 +1,5 @@
-// Copyright 2020 The Go Language Server Authors.
 // SPDX-License-Identifier: BSD-3-Clause
+// SPDX-FileCopyrightText: Copyright 2019 The Go Language Server Authors
 
 // +build !gojay
 
@@ -12,17 +12,31 @@ import (
 	"sync/atomic"
 
 	json "github.com/goccy/go-json"
+
+	"go.lsp.dev/pkg/event"
+	"go.lsp.dev/pkg/event/tag"
 )
 
 // Call implemens Conn.
-func (c *conn) Call(ctx context.Context, method string, params, result interface{}) (ID, error) {
+func (c *conn) Call(ctx context.Context, method string, params, result interface{}) (_ ID, err error) {
 	// generate a new request identifier
 	id := ID{number: atomic.AddInt64(&c.seq, 1)}
-	call, err := NewRequest(id, method, params)
+	call, err := NewCall(id, method, params)
 	if err != nil {
 		return id, fmt.Errorf("marshaling call parameters: %w", err)
 	}
 
+	ctx, done := event.Start(ctx, method,
+		tag.Method.Of(method),
+		tag.RPCDirection.Of(tag.Outbound),
+		tag.RPCID.Of(fmt.Sprintf("%q", id)),
+	)
+	defer func() {
+		recordStatus(ctx, err)
+		done()
+	}()
+	event.Metric(ctx, tag.Started.Of(1))
+
 	// We have to add ourselves to the pending map before we send, otherwise we
 	// are racing the response. Also add a buffer to rchan, so that if we get a
 	// wire response between the time this call is cancelled and id is deleted
@@ -40,7 +54,8 @@ func (c *conn) Call(ctx context.Context, method string, params, result interface
 	}()
 
 	// now we are ready to send
-	_, err = c.write(ctx, call)
+	n, err := c.write(ctx, call)
+	event.Metric(ctx, tag.SentBytes.Of(n))
 	if err != nil {
 		// sending failed, we will never get a response, so don't leave it pending
 		return id, err
@@ -50,7 +65,7 @@ func (c *conn) Call(ctx context.Context, method string, params, result interface
 	select {
 	case response := <-rchan:
 		switch {
-		case response.err != nil:
+		case response.err != nil: // is it an error response?
 			return id, response.err
 
 		case result == nil || len(response.result) == 0:

errors.go

@@ -1,5 +1,5 @@
-// Copyright 2019 The Go Language Server Authors.
 // SPDX-License-Identifier: BSD-3-Clause
+// SPDX-FileCopyrightText: Copyright 2019 The Go Language Server Authors
 
 package jsonrpc2
 
@@ -29,7 +29,6 @@ func (e *Error) Error() string {
 	if e == nil {
 		return ""
 	}
-
 	return e.Message
 }
 
@@ -48,12 +47,10 @@ func NewError(c Code, message string) *Error {
 
 // Errorf builds a Error struct for the suppied code, format and args.
 func Errorf(c Code, format string, args ...interface{}) *Error {
-	e := &Error{
+	return &Error{
 		Code:    c,
 		Message: fmt.Sprintf(format, args...),
 	}
-
-	return e
 }
 
 // constErr represents a error constant.

fake/fake.go

@@ -73,7 +73,7 @@ func NewTCPServer(ctx context.Context, server jsonrpc2.StreamServer, framer json
 		panic(fmt.Sprintf("servertest: failed to listen: %v", err))
 	}
 	if framer == nil {
-		framer = jsonrpc2.NewHeaderStream
+		framer = jsonrpc2.NewStream
 	}
 
 	go jsonrpc2.Serve(ctx, ln, server, 0)

fake/fake_test.go

@@ -16,7 +16,7 @@ type msg struct {
 	Msg string
 }
 
-func fakeHandler(ctx context.Context, reply jsonrpc2.Replier, req jsonrpc2.Requester) error {
+func fakeHandler(ctx context.Context, reply jsonrpc2.Replier, req jsonrpc2.Request) error {
 	return reply(ctx, &msg{"pong"}, nil)
 }
 

go.mod

@@ -6,4 +6,5 @@ require (
 	github.com/francoispqt/gojay v1.2.13
 	github.com/goccy/go-json v0.3.3
 	github.com/google/go-cmp v0.5.4
+	go.lsp.dev/pkg v0.0.0-20210124221122-cff50a23c8bc
 )

go.sum

@@ -101,6 +101,8 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf
 github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
 github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
 github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
+go.lsp.dev/pkg v0.0.0-20210124221122-cff50a23c8bc h1:Kzr4G5KcZrGNOU9yDm0AGeOqkzaLP9KfQVRGKHtyBog=
+go.lsp.dev/pkg v0.0.0-20210124221122-cff50a23c8bc/go.mod h1:gtSHRuYfbCT0qnbLnovpie/WEmqyJ7T4n6VXiFMBtcw=
 go.opencensus.io v0.18.0/go.mod h1:vKdFvxhtzZ9onBp9VKHK8z/sRpBMnKAsufL7wlDrCOA=
 go4.org v0.0.0-20180809161055-417644f6feb5/go.mod h1:MkTOUMDaeVYJUOUsaDXIhWPZYa1yOyC1qaOBpL57BhE=
 golang.org/x/build v0.0.0-20190111050920-041ab4dc3f9d/go.mod h1:OWs+y06UdEOHN4y+MfF/py+xQ/tYqIWW03b70/CG9Rw=

handler.go

@@ -1,18 +1,20 @@
-// Copyright 2020 The Go Language Server Authors.
 // SPDX-License-Identifier: BSD-3-Clause
+// SPDX-FileCopyrightText: Copyright 2019 The Go Language Server Authors
 
 package jsonrpc2
 
 import (
 	"context"
 	"fmt"
 	"sync"
+
+	"go.lsp.dev/pkg/event"
 )
 
 // Handler is invoked to handle incoming requests.
 //
 // The Replier sends a reply to the request and must be called exactly once.
-type Handler func(ctx context.Context, reply Replier, req Requester) error
+type Handler func(ctx context.Context, reply Replier, req Request) error
 
 // Replier is passed to handlers to allow them to reply to the request.
 //
@@ -23,14 +25,14 @@ type Replier func(ctx context.Context, result interface{}, err error) error
 // standard method not found response.
 //
 // This should normally be the final handler in a chain.
-func MethodNotFoundHandler(ctx context.Context, reply Replier, req Requester) error {
+func MethodNotFoundHandler(ctx context.Context, reply Replier, req Request) error {
 	return reply(ctx, nil, fmt.Errorf("%w: %q", ErrMethodNotFound, req.Method()))
 }
 
 // ReplyHandler creates a Handler that panics if the wrapped handler does
 // not call Reply for every request that it is passed.
 func ReplyHandler(handler Handler) Handler {
-	return func(ctx context.Context, reply Replier, req Requester) error {
+	return func(ctx context.Context, reply Replier, req Request) error {
 		called := false
 		err := handler(ctx, func(ctx context.Context, result interface{}, err error) error {
 			if called {
@@ -51,8 +53,9 @@ func ReplyHandler(handler Handler) Handler {
 func CancelHandler(handler Handler) (h Handler, cancel func(id ID)) {
 	var mu sync.Mutex
 	handling := make(map[ID]context.CancelFunc)
-	h = func(ctx context.Context, reply Replier, req Requester) error {
-		if call, ok := req.(*Request); ok {
+
+	h = func(ctx context.Context, reply Replier, req Request) error {
+		if call, ok := req.(*Call); ok {
 			cancelCtx, cancel := context.WithCancel(ctx)
 			ctx = cancelCtx
 			mu.Lock()
@@ -66,8 +69,10 @@ func CancelHandler(handler Handler) (h Handler, cancel func(id ID)) {
 				return innerReply(ctx, result, err)
 			}
 		}
+
 		return handler(ctx, reply, req)
 	}
+
 	return h, func(id ID) {
 		mu.Lock()
 		cancel, found := handling[id]
@@ -89,7 +94,8 @@ func CancelHandler(handler Handler) (h Handler, cancel func(id ID)) {
 func AsyncHandler(handler Handler) Handler {
 	nextRequest := make(chan struct{})
 	close(nextRequest)
-	return func(ctx context.Context, reply Replier, req Requester) error {
+
+	return func(ctx context.Context, reply Replier, req Request) error {
 		waitForPrevious := nextRequest
 		nextRequest = make(chan struct{})
 		unlockNext := nextRequest
@@ -98,9 +104,13 @@ func AsyncHandler(handler Handler) Handler {
 			close(unlockNext)
 			return innerReply(ctx, result, err)
 		}
+		_, queueDone := event.Start(ctx, "queued")
 		go func() {
 			<-waitForPrevious
-			handler(ctx, reply, req)
+			queueDone()
+			if err := handler(ctx, reply, req); err != nil {
+				event.Error(ctx, "jsonrpc2 async message delivery failed", err)
+			}
 		}()
 		return nil
 	}

jsonrpc2.go

@@ -1,4 +1,4 @@
-// Copyright 2019 The Go Language Server Authors.
 // SPDX-License-Identifier: BSD-3-Clause
+// SPDX-FileCopyrightText: Copyright 2019 The Go Language Server Authors
 
 package jsonrpc2