1// Copyright 2016 The go-ethereum Authors 2// This file is part of the go-ethereum library. 3// 4// The go-ethereum library is free software: you can redistribute it and/or modify 5// it under the terms of the GNU Lesser General Public License as published by 6// the Free Software Foundation, either version 3 of the License, or 7// (at your option) any later version. 8// 9// The go-ethereum library is distributed in the hope that it will be useful, 10// but WITHOUT ANY WARRANTY; without even the implied warranty of 11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 12// GNU Lesser General Public License for more details. 13// 14// You should have received a copy of the GNU Lesser General Public License 15// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. 16 17package rpc 18 19import ( 20 "context" 21 "encoding/json" 22 "errors" 23 "fmt" 24 "net/url" 25 "reflect" 26 "strconv" 27 "sync/atomic" 28 "time" 29 30 "github.com/ethereum/go-ethereum/log" 31) 32 33var ( 34 ErrClientQuit = errors.New("client is closed") 35 ErrNoResult = errors.New("no result in JSON-RPC response") 36 ErrSubscriptionQueueOverflow = errors.New("subscription queue overflow") 37 errClientReconnected = errors.New("client reconnected") 38 errDead = errors.New("connection lost") 39) 40 41const ( 42 // Timeouts 43 defaultDialTimeout = 10 * time.Second // used if context has no deadline 44 subscribeTimeout = 5 * time.Second // overall timeout eth_subscribe, rpc_modules calls 45) 46 47const ( 48 // Subscriptions are removed when the subscriber cannot keep up. 49 // 50 // This can be worked around by supplying a channel with sufficiently sized buffer, 51 // but this can be inconvenient and hard to explain in the docs. Another issue with 52 // buffered channels is that the buffer is static even though it might not be needed 53 // most of the time. 54 // 55 // The approach taken here is to maintain a per-subscription linked list buffer 56 // shrinks on demand. If the buffer reaches the size below, the subscription is 57 // dropped. 58 maxClientSubscriptionBuffer = 20000 59) 60 61const ( 62 httpScheme = "http" 63 wsScheme = "ws" 64 ipcScheme = "ipc" 65) 66 67// BatchElem is an element in a batch request. 68type BatchElem struct { 69 Method string 70 Args []interface{} 71 // The result is unmarshaled into this field. Result must be set to a 72 // non-nil pointer value of the desired type, otherwise the response will be 73 // discarded. 74 Result interface{} 75 // Error is set if the server returns an error for this request, or if 76 // unmarshaling into Result fails. It is not set for I/O errors. 77 Error error 78} 79 80// Client represents a connection to an RPC server. 81type Client struct { 82 idgen func() ID // for subscriptions 83 scheme string // connection type: http, ws or ipc 84 services *serviceRegistry 85 86 idCounter uint32 87 88 // This function, if non-nil, is called when the connection is lost. 89 reconnectFunc reconnectFunc 90 91 // writeConn is used for writing to the connection on the caller's goroutine. It should 92 // only be accessed outside of dispatch, with the write lock held. The write lock is 93 // taken by sending on reqInit and released by sending on reqSent. 94 writeConn jsonWriter 95 96 // for dispatch 97 close chan struct{} 98 closing chan struct{} // closed when client is quitting 99 didClose chan struct{} // closed when client quits 100 reconnected chan ServerCodec // where write/reconnect sends the new connection 101 readOp chan readOp // read messages 102 readErr chan error // errors from read 103 reqInit chan *requestOp // register response IDs, takes write lock 104 reqSent chan error // signals write completion, releases write lock 105 reqTimeout chan *requestOp // removes response IDs when call timeout expires 106} 107 108type reconnectFunc func(ctx context.Context) (ServerCodec, error) 109 110type clientContextKey struct{} 111 112type clientConn struct { 113 codec ServerCodec 114 handler *handler 115} 116 117func (c *Client) newClientConn(conn ServerCodec) *clientConn { 118 ctx := context.WithValue(context.Background(), clientContextKey{}, c) 119 // Http connections have already set the scheme 120 if !c.isHTTP() && c.scheme != "" { 121 ctx = context.WithValue(ctx, "scheme", c.scheme) 122 } 123 handler := newHandler(ctx, conn, c.idgen, c.services) 124 return &clientConn{conn, handler} 125} 126 127func (cc *clientConn) close(err error, inflightReq *requestOp) { 128 cc.handler.close(err, inflightReq) 129 cc.codec.close() 130} 131 132type readOp struct { 133 msgs []*jsonrpcMessage 134 batch bool 135} 136 137type requestOp struct { 138 ids []json.RawMessage 139 err error 140 resp chan *jsonrpcMessage // receives up to len(ids) responses 141 sub *ClientSubscription // only set for EthSubscribe requests 142} 143 144func (op *requestOp) wait(ctx context.Context, c *Client) (*jsonrpcMessage, error) { 145 select { 146 case <-ctx.Done(): 147 // Send the timeout to dispatch so it can remove the request IDs. 148 if !c.isHTTP() { 149 select { 150 case c.reqTimeout <- op: 151 case <-c.closing: 152 } 153 } 154 return nil, ctx.Err() 155 case resp := <-op.resp: 156 return resp, op.err 157 } 158} 159 160// Dial creates a new client for the given URL. 161// 162// The currently supported URL schemes are "http", "https", "ws" and "wss". If rawurl is a 163// file name with no URL scheme, a local socket connection is established using UNIX 164// domain sockets on supported platforms and named pipes on Windows. If you want to 165// configure transport options, use DialHTTP, DialWebsocket or DialIPC instead. 166// 167// For websocket connections, the origin is set to the local host name. 168// 169// The client reconnects automatically if the connection is lost. 170func Dial(rawurl string) (*Client, error) { 171 return DialContext(context.Background(), rawurl) 172} 173 174// DialContext creates a new RPC client, just like Dial. 175// 176// The context is used to cancel or time out the initial connection establishment. It does 177// not affect subsequent interactions with the client. 178func DialContext(ctx context.Context, rawurl string) (*Client, error) { 179 u, err := url.Parse(rawurl) 180 if err != nil { 181 return nil, err 182 } 183 switch u.Scheme { 184 case "http", "https": 185 return DialHTTP(rawurl) 186 case "ws", "wss": 187 return DialWebsocket(ctx, rawurl, "") 188 case "stdio": 189 return DialStdIO(ctx) 190 case "": 191 return DialIPC(ctx, rawurl) 192 default: 193 return nil, fmt.Errorf("no known transport for URL scheme %q", u.Scheme) 194 } 195} 196 197// Client retrieves the client from the context, if any. This can be used to perform 198// 'reverse calls' in a handler method. 199func ClientFromContext(ctx context.Context) (*Client, bool) { 200 client, ok := ctx.Value(clientContextKey{}).(*Client) 201 return client, ok 202} 203 204func newClient(initctx context.Context, connect reconnectFunc) (*Client, error) { 205 conn, err := connect(initctx) 206 if err != nil { 207 return nil, err 208 } 209 c := initClient(conn, randomIDGenerator(), new(serviceRegistry)) 210 c.reconnectFunc = connect 211 return c, nil 212} 213 214func initClient(conn ServerCodec, idgen func() ID, services *serviceRegistry) *Client { 215 scheme := "" 216 switch conn.(type) { 217 case *httpConn: 218 scheme = httpScheme 219 case *websocketCodec: 220 scheme = wsScheme 221 case *jsonCodec: 222 scheme = ipcScheme 223 } 224 c := &Client{ 225 idgen: idgen, 226 scheme: scheme, 227 services: services, 228 writeConn: conn, 229 close: make(chan struct{}), 230 closing: make(chan struct{}), 231 didClose: make(chan struct{}), 232 reconnected: make(chan ServerCodec), 233 readOp: make(chan readOp), 234 readErr: make(chan error), 235 reqInit: make(chan *requestOp), 236 reqSent: make(chan error, 1), 237 reqTimeout: make(chan *requestOp), 238 } 239 if !c.isHTTP() { 240 go c.dispatch(conn) 241 } 242 return c 243} 244 245// RegisterName creates a service for the given receiver type under the given name. When no 246// methods on the given receiver match the criteria to be either a RPC method or a 247// subscription an error is returned. Otherwise a new service is created and added to the 248// service collection this client provides to the server. 249func (c *Client) RegisterName(name string, receiver interface{}) error { 250 return c.services.registerName(name, receiver) 251} 252 253func (c *Client) nextID() json.RawMessage { 254 id := atomic.AddUint32(&c.idCounter, 1) 255 return strconv.AppendUint(nil, uint64(id), 10) 256} 257 258// SupportedModules calls the rpc_modules method, retrieving the list of 259// APIs that are available on the server. 260func (c *Client) SupportedModules() (map[string]string, error) { 261 var result map[string]string 262 ctx, cancel := context.WithTimeout(context.Background(), subscribeTimeout) 263 defer cancel() 264 err := c.CallContext(ctx, &result, "rpc_modules") 265 return result, err 266} 267 268// Close closes the client, aborting any in-flight requests. 269func (c *Client) Close() { 270 if c.isHTTP() { 271 return 272 } 273 select { 274 case c.close <- struct{}{}: 275 <-c.didClose 276 case <-c.didClose: 277 } 278} 279 280// SetHeader adds a custom HTTP header to the client's requests. 281// This method only works for clients using HTTP, it doesn't have 282// any effect for clients using another transport. 283func (c *Client) SetHeader(key, value string) { 284 if !c.isHTTP() { 285 return 286 } 287 conn := c.writeConn.(*httpConn) 288 conn.mu.Lock() 289 conn.headers.Set(key, value) 290 conn.mu.Unlock() 291} 292 293// Call performs a JSON-RPC call with the given arguments and unmarshals into 294// result if no error occurred. 295// 296// The result must be a pointer so that package json can unmarshal into it. You 297// can also pass nil, in which case the result is ignored. 298func (c *Client) Call(result interface{}, method string, args ...interface{}) error { 299 ctx := context.Background() 300 return c.CallContext(ctx, result, method, args...) 301} 302 303// CallContext performs a JSON-RPC call with the given arguments. If the context is 304// canceled before the call has successfully returned, CallContext returns immediately. 305// 306// The result must be a pointer so that package json can unmarshal into it. You 307// can also pass nil, in which case the result is ignored. 308func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error { 309 if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr { 310 return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result) 311 } 312 msg, err := c.newMessage(method, args...) 313 if err != nil { 314 return err 315 } 316 op := &requestOp{ids: []json.RawMessage{msg.ID}, resp: make(chan *jsonrpcMessage, 1)} 317 318 if c.isHTTP() { 319 err = c.sendHTTP(ctx, op, msg) 320 } else { 321 err = c.send(ctx, op, msg) 322 } 323 if err != nil { 324 return err 325 } 326 327 // dispatch has accepted the request and will close the channel when it quits. 328 switch resp, err := op.wait(ctx, c); { 329 case err != nil: 330 return err 331 case resp.Error != nil: 332 return resp.Error 333 case len(resp.Result) == 0: 334 return ErrNoResult 335 default: 336 return json.Unmarshal(resp.Result, &result) 337 } 338} 339 340// BatchCall sends all given requests as a single batch and waits for the server 341// to return a response for all of them. 342// 343// In contrast to Call, BatchCall only returns I/O errors. Any error specific to 344// a request is reported through the Error field of the corresponding BatchElem. 345// 346// Note that batch calls may not be executed atomically on the server side. 347func (c *Client) BatchCall(b []BatchElem) error { 348 ctx := context.Background() 349 return c.BatchCallContext(ctx, b) 350} 351 352// BatchCall sends all given requests as a single batch and waits for the server 353// to return a response for all of them. The wait duration is bounded by the 354// context's deadline. 355// 356// In contrast to CallContext, BatchCallContext only returns errors that have occurred 357// while sending the request. Any error specific to a request is reported through the 358// Error field of the corresponding BatchElem. 359// 360// Note that batch calls may not be executed atomically on the server side. 361func (c *Client) BatchCallContext(ctx context.Context, b []BatchElem) error { 362 var ( 363 msgs = make([]*jsonrpcMessage, len(b)) 364 byID = make(map[string]int, len(b)) 365 ) 366 op := &requestOp{ 367 ids: make([]json.RawMessage, len(b)), 368 resp: make(chan *jsonrpcMessage, len(b)), 369 } 370 for i, elem := range b { 371 msg, err := c.newMessage(elem.Method, elem.Args...) 372 if err != nil { 373 return err 374 } 375 msgs[i] = msg 376 op.ids[i] = msg.ID 377 byID[string(msg.ID)] = i 378 } 379 380 var err error 381 if c.isHTTP() { 382 err = c.sendBatchHTTP(ctx, op, msgs) 383 } else { 384 err = c.send(ctx, op, msgs) 385 } 386 387 // Wait for all responses to come back. 388 for n := 0; n < len(b) && err == nil; n++ { 389 var resp *jsonrpcMessage 390 resp, err = op.wait(ctx, c) 391 if err != nil { 392 break 393 } 394 // Find the element corresponding to this response. 395 // The element is guaranteed to be present because dispatch 396 // only sends valid IDs to our channel. 397 elem := &b[byID[string(resp.ID)]] 398 if resp.Error != nil { 399 elem.Error = resp.Error 400 continue 401 } 402 if len(resp.Result) == 0 { 403 elem.Error = ErrNoResult 404 continue 405 } 406 elem.Error = json.Unmarshal(resp.Result, elem.Result) 407 } 408 return err 409} 410 411// Notify sends a notification, i.e. a method call that doesn't expect a response. 412func (c *Client) Notify(ctx context.Context, method string, args ...interface{}) error { 413 op := new(requestOp) 414 msg, err := c.newMessage(method, args...) 415 if err != nil { 416 return err 417 } 418 msg.ID = nil 419 420 if c.isHTTP() { 421 return c.sendHTTP(ctx, op, msg) 422 } 423 return c.send(ctx, op, msg) 424} 425 426// EthSubscribe registers a subscription under the "eth" namespace. 427func (c *Client) EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error) { 428 return c.Subscribe(ctx, "eth", channel, args...) 429} 430 431// ShhSubscribe registers a subscription under the "shh" namespace. 432// Deprecated: use Subscribe(ctx, "shh", ...). 433func (c *Client) ShhSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error) { 434 return c.Subscribe(ctx, "shh", channel, args...) 435} 436 437// Subscribe calls the "<namespace>_subscribe" method with the given arguments, 438// registering a subscription. Server notifications for the subscription are 439// sent to the given channel. The element type of the channel must match the 440// expected type of content returned by the subscription. 441// 442// The context argument cancels the RPC request that sets up the subscription but has no 443// effect on the subscription after Subscribe has returned. 444// 445// Slow subscribers will be dropped eventually. Client buffers up to 20000 notifications 446// before considering the subscriber dead. The subscription Err channel will receive 447// ErrSubscriptionQueueOverflow. Use a sufficiently large buffer on the channel or ensure 448// that the channel usually has at least one reader to prevent this issue. 449func (c *Client) Subscribe(ctx context.Context, namespace string, channel interface{}, args ...interface{}) (*ClientSubscription, error) { 450 // Check type of channel first. 451 chanVal := reflect.ValueOf(channel) 452 if chanVal.Kind() != reflect.Chan || chanVal.Type().ChanDir()&reflect.SendDir == 0 { 453 panic("first argument to Subscribe must be a writable channel") 454 } 455 if chanVal.IsNil() { 456 panic("channel given to Subscribe must not be nil") 457 } 458 if c.isHTTP() { 459 return nil, ErrNotificationsUnsupported 460 } 461 462 msg, err := c.newMessage(namespace+subscribeMethodSuffix, args...) 463 if err != nil { 464 return nil, err 465 } 466 op := &requestOp{ 467 ids: []json.RawMessage{msg.ID}, 468 resp: make(chan *jsonrpcMessage), 469 sub: newClientSubscription(c, namespace, chanVal), 470 } 471 472 // Send the subscription request. 473 // The arrival and validity of the response is signaled on sub.quit. 474 if err := c.send(ctx, op, msg); err != nil { 475 return nil, err 476 } 477 if _, err := op.wait(ctx, c); err != nil { 478 return nil, err 479 } 480 return op.sub, nil 481} 482 483func (c *Client) newMessage(method string, paramsIn ...interface{}) (*jsonrpcMessage, error) { 484 msg := &jsonrpcMessage{Version: vsn, ID: c.nextID(), Method: method} 485 if paramsIn != nil { // prevent sending "params":null 486 var err error 487 if msg.Params, err = json.Marshal(paramsIn); err != nil { 488 return nil, err 489 } 490 } 491 return msg, nil 492} 493 494// send registers op with the dispatch loop, then sends msg on the connection. 495// if sending fails, op is deregistered. 496func (c *Client) send(ctx context.Context, op *requestOp, msg interface{}) error { 497 select { 498 case c.reqInit <- op: 499 err := c.write(ctx, msg, false) 500 c.reqSent <- err 501 return err 502 case <-ctx.Done(): 503 // This can happen if the client is overloaded or unable to keep up with 504 // subscription notifications. 505 return ctx.Err() 506 case <-c.closing: 507 return ErrClientQuit 508 } 509} 510 511func (c *Client) write(ctx context.Context, msg interface{}, retry bool) error { 512 // The previous write failed. Try to establish a new connection. 513 if c.writeConn == nil { 514 if err := c.reconnect(ctx); err != nil { 515 return err 516 } 517 } 518 err := c.writeConn.writeJSON(ctx, msg) 519 if err != nil { 520 c.writeConn = nil 521 if !retry { 522 return c.write(ctx, msg, true) 523 } 524 } 525 return err 526} 527 528func (c *Client) reconnect(ctx context.Context) error { 529 if c.reconnectFunc == nil { 530 return errDead 531 } 532 533 if _, ok := ctx.Deadline(); !ok { 534 var cancel func() 535 ctx, cancel = context.WithTimeout(ctx, defaultDialTimeout) 536 defer cancel() 537 } 538 newconn, err := c.reconnectFunc(ctx) 539 if err != nil { 540 log.Trace("RPC client reconnect failed", "err", err) 541 return err 542 } 543 select { 544 case c.reconnected <- newconn: 545 c.writeConn = newconn 546 return nil 547 case <-c.didClose: 548 newconn.close() 549 return ErrClientQuit 550 } 551} 552 553// dispatch is the main loop of the client. 554// It sends read messages to waiting calls to Call and BatchCall 555// and subscription notifications to registered subscriptions. 556func (c *Client) dispatch(codec ServerCodec) { 557 var ( 558 lastOp *requestOp // tracks last send operation 559 reqInitLock = c.reqInit // nil while the send lock is held 560 conn = c.newClientConn(codec) 561 reading = true 562 ) 563 defer func() { 564 close(c.closing) 565 if reading { 566 conn.close(ErrClientQuit, nil) 567 c.drainRead() 568 } 569 close(c.didClose) 570 }() 571 572 // Spawn the initial read loop. 573 go c.read(codec) 574 575 for { 576 select { 577 case <-c.close: 578 return 579 580 // Read path: 581 case op := <-c.readOp: 582 if op.batch { 583 conn.handler.handleBatch(op.msgs) 584 } else { 585 conn.handler.handleMsg(op.msgs[0]) 586 } 587 588 case err := <-c.readErr: 589 conn.handler.log.Debug("RPC connection read error", "err", err) 590 conn.close(err, lastOp) 591 reading = false 592 593 // Reconnect: 594 case newcodec := <-c.reconnected: 595 log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.remoteAddr()) 596 if reading { 597 // Wait for the previous read loop to exit. This is a rare case which 598 // happens if this loop isn't notified in time after the connection breaks. 599 // In those cases the caller will notice first and reconnect. Closing the 600 // handler terminates all waiting requests (closing op.resp) except for 601 // lastOp, which will be transferred to the new handler. 602 conn.close(errClientReconnected, lastOp) 603 c.drainRead() 604 } 605 go c.read(newcodec) 606 reading = true 607 conn = c.newClientConn(newcodec) 608 // Re-register the in-flight request on the new handler 609 // because that's where it will be sent. 610 conn.handler.addRequestOp(lastOp) 611 612 // Send path: 613 case op := <-reqInitLock: 614 // Stop listening for further requests until the current one has been sent. 615 reqInitLock = nil 616 lastOp = op 617 conn.handler.addRequestOp(op) 618 619 case err := <-c.reqSent: 620 if err != nil { 621 // Remove response handlers for the last send. When the read loop 622 // goes down, it will signal all other current operations. 623 conn.handler.removeRequestOp(lastOp) 624 } 625 // Let the next request in. 626 reqInitLock = c.reqInit 627 lastOp = nil 628 629 case op := <-c.reqTimeout: 630 conn.handler.removeRequestOp(op) 631 } 632 } 633} 634 635// drainRead drops read messages until an error occurs. 636func (c *Client) drainRead() { 637 for { 638 select { 639 case <-c.readOp: 640 case <-c.readErr: 641 return 642 } 643 } 644} 645 646// read decodes RPC messages from a codec, feeding them into dispatch. 647func (c *Client) read(codec ServerCodec) { 648 for { 649 msgs, batch, err := codec.readBatch() 650 if _, ok := err.(*json.SyntaxError); ok { 651 codec.writeJSON(context.Background(), errorMessage(&parseError{err.Error()})) 652 } 653 if err != nil { 654 c.readErr <- err 655 return 656 } 657 c.readOp <- readOp{msgs, batch} 658 } 659} 660 661func (c *Client) isHTTP() bool { 662 return c.scheme == httpScheme 663} 664