diff options
Diffstat (limited to 'rpc/client.go')
| -rw-r--r-- | rpc/client.go | 39 |
1 files changed, 29 insertions, 10 deletions
diff --git a/rpc/client.go b/rpc/client.go index 1c7058b..4f36a05 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -28,7 +28,7 @@ import ( "sync/atomic" "time" - "github.com/ava-labs/go-ethereum/log" + "github.com/ethereum/go-ethereum/log" ) var ( @@ -85,7 +85,7 @@ type Client struct { // writeConn is used for writing to the connection on the caller's goroutine. It should // only be accessed outside of dispatch, with the write lock held. The write lock is - // taken by sending on requestOp and released by sending on sendDone. + // taken by sending on reqInit and released by sending on reqSent. writeConn jsonWriter // for dispatch @@ -117,7 +117,7 @@ func (c *Client) newClientConn(conn ServerCodec) *clientConn { func (cc *clientConn) close(err error, inflightReq *requestOp) { cc.handler.close(err, inflightReq) - cc.codec.Close() + cc.codec.close() } type readOp struct { @@ -260,6 +260,19 @@ func (c *Client) Close() { } } +// SetHeader adds a custom HTTP header to the client's requests. +// This method only works for clients using HTTP, it doesn't have +// any effect for clients using another transport. +func (c *Client) SetHeader(key, value string) { + if !c.isHTTP { + return + } + conn := c.writeConn.(*httpConn) + conn.mu.Lock() + conn.headers.Set(key, value) + conn.mu.Unlock() +} + // Call performs a JSON-RPC call with the given arguments and unmarshals into // result if no error occurred. // @@ -276,6 +289,9 @@ func (c *Client) Call(result interface{}, method string, args ...interface{}) er // The result must be a pointer so that package json can unmarshal into it. You // can also pass nil, in which case the result is ignored. func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error { + if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr { + return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result) + } msg, err := c.newMessage(method, args...) if err != nil { return err @@ -465,7 +481,7 @@ func (c *Client) newMessage(method string, paramsIn ...interface{}) (*jsonrpcMes func (c *Client) send(ctx context.Context, op *requestOp, msg interface{}) error { select { case c.reqInit <- op: - err := c.write(ctx, msg) + err := c.write(ctx, msg, false) c.reqSent <- err return err case <-ctx.Done(): @@ -477,16 +493,19 @@ func (c *Client) send(ctx context.Context, op *requestOp, msg interface{}) error } } -func (c *Client) write(ctx context.Context, msg interface{}) error { +func (c *Client) write(ctx context.Context, msg interface{}, retry bool) error { // The previous write failed. Try to establish a new connection. if c.writeConn == nil { if err := c.reconnect(ctx); err != nil { return err } } - err := c.writeConn.Write(ctx, msg) + err := c.writeConn.writeJSON(ctx, msg) if err != nil { c.writeConn = nil + if !retry { + return c.write(ctx, msg, true) + } } return err } @@ -511,7 +530,7 @@ func (c *Client) reconnect(ctx context.Context) error { c.writeConn = newconn return nil case <-c.didClose: - newconn.Close() + newconn.close() return ErrClientQuit } } @@ -558,7 +577,7 @@ func (c *Client) dispatch(codec ServerCodec) { // Reconnect: case newcodec := <-c.reconnected: - log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.RemoteAddr()) + log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.remoteAddr()) if reading { // Wait for the previous read loop to exit. This is a rare case which // happens if this loop isn't notified in time after the connection breaks. @@ -612,9 +631,9 @@ func (c *Client) drainRead() { // read decodes RPC messages from a codec, feeding them into dispatch. func (c *Client) read(codec ServerCodec) { for { - msgs, batch, err := codec.Read() + msgs, batch, err := codec.readBatch() if _, ok := err.(*json.SyntaxError); ok { - codec.Write(context.Background(), errorMessage(&parseError{err.Error()})) + codec.writeJSON(context.Background(), errorMessage(&parseError{err.Error()})) } if err != nil { c.readErr <- err |