1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313
|
// Copyright 2018 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package protocol
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"cuelang.org/go/internal/golangorgx/gopls/telemetry"
"cuelang.org/go/internal/golangorgx/gopls/util/bug"
"cuelang.org/go/internal/golangorgx/tools/event"
"cuelang.org/go/internal/golangorgx/tools/jsonrpc2"
jsonrpc2_v2 "cuelang.org/go/internal/golangorgx/tools/jsonrpc2_v2"
"cuelang.org/go/internal/golangorgx/tools/xcontext"
)
var (
// RequestCancelledError should be used when a request is cancelled early.
RequestCancelledError = jsonrpc2.NewError(-32800, "JSON RPC cancelled")
RequestCancelledErrorV2 = jsonrpc2_v2.NewError(-32800, "JSON RPC cancelled")
)
type ClientCloser interface {
Client
io.Closer
}
type connSender interface {
io.Closer
Notify(ctx context.Context, method string, params interface{}) error
Call(ctx context.Context, method string, params, result interface{}) error
}
type clientDispatcher struct {
sender connSender
}
func (c *clientDispatcher) Close() error {
return c.sender.Close()
}
// ClientDispatcher returns a Client that dispatches LSP requests across the
// given jsonrpc2 connection.
func ClientDispatcher(conn jsonrpc2.Conn) ClientCloser {
return &clientDispatcher{sender: clientConn{conn}}
}
type clientConn struct {
conn jsonrpc2.Conn
}
func (c clientConn) Close() error {
return c.conn.Close()
}
func (c clientConn) Notify(ctx context.Context, method string, params interface{}) error {
return c.conn.Notify(ctx, method, params)
}
func (c clientConn) Call(ctx context.Context, method string, params interface{}, result interface{}) error {
id, err := c.conn.Call(ctx, method, params, result)
if ctx.Err() != nil {
cancelCall(ctx, c, id)
}
return err
}
func ClientDispatcherV2(conn *jsonrpc2_v2.Connection) ClientCloser {
return &clientDispatcher{clientConnV2{conn}}
}
type clientConnV2 struct {
conn *jsonrpc2_v2.Connection
}
func (c clientConnV2) Close() error {
return c.conn.Close()
}
func (c clientConnV2) Notify(ctx context.Context, method string, params interface{}) error {
return c.conn.Notify(ctx, method, params)
}
func (c clientConnV2) Call(ctx context.Context, method string, params interface{}, result interface{}) error {
call := c.conn.Call(ctx, method, params)
err := call.Await(ctx, result)
if ctx.Err() != nil {
detached := xcontext.Detach(ctx)
c.conn.Notify(detached, "$/cancelRequest", &CancelParams{ID: call.ID().Raw()})
}
return err
}
// ServerDispatcher returns a Server that dispatches LSP requests across the
// given jsonrpc2 connection.
func ServerDispatcher(conn jsonrpc2.Conn) Server {
return &serverDispatcher{sender: clientConn{conn}}
}
func ServerDispatcherV2(conn *jsonrpc2_v2.Connection) Server {
return &serverDispatcher{sender: clientConnV2{conn}}
}
type serverDispatcher struct {
sender connSender
}
func ClientHandler(client Client, handler jsonrpc2.Handler) jsonrpc2.Handler {
return func(ctx context.Context, reply jsonrpc2.Replier, req jsonrpc2.Request) error {
if ctx.Err() != nil {
ctx := xcontext.Detach(ctx)
return reply(ctx, nil, RequestCancelledError)
}
handled, err := clientDispatch(ctx, client, reply, req)
if handled || err != nil {
return err
}
return handler(ctx, reply, req)
}
}
func ClientHandlerV2(client Client) jsonrpc2_v2.Handler {
return jsonrpc2_v2.HandlerFunc(func(ctx context.Context, req *jsonrpc2_v2.Request) (interface{}, error) {
if ctx.Err() != nil {
return nil, RequestCancelledErrorV2
}
req1 := req2to1(req)
var (
result interface{}
resErr error
)
replier := func(_ context.Context, res interface{}, err error) error {
if err != nil {
resErr = err
return nil
}
result = res
return nil
}
_, err := clientDispatch(ctx, client, replier, req1)
if err != nil {
return nil, err
}
return result, resErr
})
}
func ServerHandler(server Server, handler jsonrpc2.Handler) jsonrpc2.Handler {
return func(ctx context.Context, reply jsonrpc2.Replier, req jsonrpc2.Request) error {
if ctx.Err() != nil {
ctx := xcontext.Detach(ctx)
return reply(ctx, nil, RequestCancelledError)
}
handled, err := serverDispatch(ctx, server, reply, req)
if handled || err != nil {
return err
}
return handler(ctx, reply, req)
}
}
func ServerHandlerV2(server Server) jsonrpc2_v2.Handler {
return jsonrpc2_v2.HandlerFunc(func(ctx context.Context, req *jsonrpc2_v2.Request) (interface{}, error) {
if ctx.Err() != nil {
return nil, RequestCancelledErrorV2
}
req1 := req2to1(req)
var (
result interface{}
resErr error
)
replier := func(_ context.Context, res interface{}, err error) error {
if err != nil {
resErr = err
return nil
}
result = res
return nil
}
_, err := serverDispatch(ctx, server, replier, req1)
if err != nil {
return nil, err
}
return result, resErr
})
}
func req2to1(req2 *jsonrpc2_v2.Request) jsonrpc2.Request {
if req2.ID.IsValid() {
raw := req2.ID.Raw()
var idv1 jsonrpc2.ID
switch v := raw.(type) {
case int64:
idv1 = jsonrpc2.NewIntID(v)
case string:
idv1 = jsonrpc2.NewStringID(v)
default:
panic(fmt.Sprintf("unsupported ID type %T", raw))
}
req1, err := jsonrpc2.NewCall(idv1, req2.Method, req2.Params)
if err != nil {
panic(err)
}
return req1
}
req1, err := jsonrpc2.NewNotification(req2.Method, req2.Params)
if err != nil {
panic(err)
}
return req1
}
func Handlers(handler jsonrpc2.Handler) jsonrpc2.Handler {
return CancelHandler(
jsonrpc2.AsyncHandler(
jsonrpc2.MustReplyHandler(handler)))
}
func CancelHandler(handler jsonrpc2.Handler) jsonrpc2.Handler {
handler, canceller := jsonrpc2.CancelHandler(handler)
return func(ctx context.Context, reply jsonrpc2.Replier, req jsonrpc2.Request) error {
if req.Method() != "$/cancelRequest" {
// TODO(iancottrell): See if we can generate a reply for the request to be cancelled
// at the point of cancellation rather than waiting for gopls to naturally reply.
// To do that, we need to keep track of whether a reply has been sent already and
// be careful about racing between the two paths.
// TODO(iancottrell): Add a test that watches the stream and verifies the response
// for the cancelled request flows.
replyWithDetachedContext := func(ctx context.Context, resp interface{}, err error) error {
// https://microsoft.github.io/language-server-protocol/specifications/specification-current/#cancelRequest
if ctx.Err() != nil && err == nil {
err = RequestCancelledError
}
ctx = xcontext.Detach(ctx)
return reply(ctx, resp, err)
}
return handler(ctx, replyWithDetachedContext, req)
}
var params CancelParams
if err := UnmarshalJSON(req.Params(), ¶ms); err != nil {
return sendParseError(ctx, reply, err)
}
if n, ok := params.ID.(float64); ok {
canceller(jsonrpc2.NewIntID(int64(n)))
} else if s, ok := params.ID.(string); ok {
canceller(jsonrpc2.NewStringID(s))
} else {
return sendParseError(ctx, reply, fmt.Errorf("request ID %v malformed", params.ID))
}
return reply(ctx, nil, nil)
}
}
func Call(ctx context.Context, conn jsonrpc2.Conn, method string, params interface{}, result interface{}) error {
id, err := conn.Call(ctx, method, params, result)
if ctx.Err() != nil {
cancelCall(ctx, clientConn{conn}, id)
}
return err
}
func cancelCall(ctx context.Context, sender connSender, id jsonrpc2.ID) {
ctx = xcontext.Detach(ctx)
ctx, done := event.Start(ctx, "protocol.canceller")
defer done()
// Note that only *jsonrpc2.ID implements json.Marshaler.
sender.Notify(ctx, "$/cancelRequest", &CancelParams{ID: &id})
}
// UnmarshalJSON unmarshals msg into the variable pointed to by
// params. In JSONRPC, optional messages may be
// "null", in which case it is a no-op.
func UnmarshalJSON(msg json.RawMessage, v any) error {
if len(msg) == 0 || bytes.Equal(msg, []byte("null")) {
return nil
}
return json.Unmarshal(msg, v)
}
func sendParseError(ctx context.Context, reply jsonrpc2.Replier, err error) error {
return reply(ctx, nil, fmt.Errorf("%w: %s", jsonrpc2.ErrParse, err))
}
// NonNilSlice returns x, or an empty slice if x was nil.
//
// (Many slice fields of protocol structs must be non-nil
// to avoid being encoded as JSON "null".)
func NonNilSlice[T comparable](x []T) []T {
if x == nil {
return []T{}
}
return x
}
func recoverHandlerPanic(method string) {
// Report panics in the handler goroutine,
// unless we have enabled the monitor,
// which reports all crashes.
if !telemetry.CrashMonitorSupported() {
defer func() {
if x := recover(); x != nil {
bug.Reportf("panic in %s request", method)
panic(x)
}
}()
}
}
|