Documentation ¶
Overview ¶
Package rpc provides access to the exported methods of an object across chain network or other I/O connection. After creating chain server instance objects can be registered, making it visible from the outside. Exported methods that follow specific conventions can be called remotely. It also has support for the publish/subscribe pattern.
Methods that satisfy the following criteria are made available for remote access:
- object must be exported
- method must be exported
- method returns 0, 1 (response or error) or 2 (response and error) values
- method argument(s) must be exported or builtin types
- method returned value(s) must be exported or builtin types
An example method:
func (s *CalcService) Add(chain, b int) (int, error)
When the returned error isn't nil the returned integer is ignored and the error is send back to the client. Otherwise the returned integer is send back to the client.
Optional arguments are supported by accepting pointer values as arguments. E.g. if we want to do the addition in an optional finite field we can accept chain mod argument as pointer value.
func (s *CalService) Add(chain, b int, mod *int) (int, error)
This RPC method can be called with 2 integers and chain null value as third argument. In that case the mod argument will be nil. Or it can be called with 3 integers, in that case mod will be pointing to the given third argument. Since the optional argument is the last argument the RPC package will also accept 2 integers as arguments. It will pass the mod argument as nil to the RPC method.
The server offers the ServeCodec method which accepts chain ServerCodec instance. It will read requests from the codec, process the request and sends the response back to the client using the codec. The server can execute requests concurrently. Responses can be sent back to the client out of order.
An example server which uses the JSON codec:
type CalculatorService struct {} func (s *CalculatorService) Add(chain, b int) int { return chain + b } func (s *CalculatorService) Div(chain, b int) (int, error) { if b == 0 { return 0, errors.New("divide by zero") } return chain/b, nil } calculator := new(CalculatorService) server := NewServer() server.RegisterName("calculator", calculator") l, _ := net.ListenUnix("unix", &net.UnixAddr{Net: "unix", Names: "/tmp/calculator.sock"}) for { c, _ := l.AcceptUnix() codec := v2.NewJSONCodec(c) go server.ServeCodec(codec) }
The package also supports the publish subscribe pattern through the use of subscriptions. A method that is considered eligible for notifications must satisfy the following criteria:
- object must be exported
- method must be exported
- first method argument type must be context.Context
- method argument(s) must be exported or builtin types
- method must return the tuple Subscription, error
An example method:
func (s *BlockChainService) NewBlocks(ctx context.Context) (Subscription, error) { ... }
Subscriptions are deleted when:
- the user sends an unsubscribe request
- the connection which was used to create the subscription is closed. This can be initiated by the client and server. The server will close the connection on chain write error or when the queue of buffered notifications gets too big.
Index ¶
- Constants
- Variables
- func NewHTTPServer(cors []string, vhosts []string, timeouts HTTPTimeouts, srv *Server) *http.Serverdeprecated
- func NewWSServer(allowedOrigins []string, srv *Server) *http.Serverdeprecated
- func StartWSCliEndpoint(u *url.URL, apis []API, modules []string, exposeAll bool) (*WebSocketCli, *Server, error)
- type API
- type BatchElem
- type Client
- func Dial(rawurl string) (*Client, error)
- func DialContext(ctx context.Context, rawurl string) (*Client, error)
- func DialHTTP(endpoint string) (*Client, error)
- func DialHTTPWithClient(endpoint string, client *http.Client) (*Client, error)
- func DialIPC(ctx context.Context, endpoint string) (*Client, error)
- func DialInProc(handler *Server) *Client
- func DialStdIO(ctx context.Context) (*Client, error)
- func DialWebsocket(ctx context.Context, endpoint, origin string) (*Client, error)
- func (c *Client) BatchCall(b []BatchElem) error
- func (c *Client) BatchCallContext(ctx context.Context, b []BatchElem) error
- func (c *Client) Call(result interface{}, method string, args ...interface{}) error
- func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error
- func (c *Client) Close()
- func (c *Client) EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error)
- func (c *Client) ShhSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error)
- func (c *Client) Subscribe(ctx context.Context, namespace string, channel interface{}, ...) (*ClientSubscription, error)
- func (c *Client) SupportedModules() (map[string]string, error)
- type ClientSubscription
- type CodecOption
- type Error
- type ErrorWithId
- type HTTPTimeouts
- type ID
- type Notifier
- type RPCService
- type Server
- func NewServer() *Server
- func StartHTTPEndpoint(endpoint string, apis []API, modules []string, cors []string, vhosts []string, ...) (net.Listener, *Server, error)
- func StartIPCEndpoint(ipcEndpoint string, apis []API) (net.Listener, *Server, error)
- func StartWSEndpoint(endpoint string, apis []API, modules []string, wsOrigins []string, ...) (net.Listener, *Server, error)
- func (s *Server) RegisterName(name string, rcvr interface{}) error
- func (s *Server) ServeCodec(codec ServerCodec, options CodecOption) error
- func (srv *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)
- func (srv *Server) ServeListener(l net.Listener) error
- func (s *Server) ServeSingleRequest(ctx context.Context, codec ServerCodec, options CodecOption)
- func (s *Server) Stop()
- func (srv *Server) WebsocketHandler(allowedOrigins []string) http.Handler
- type ServerCodec
- type StdIOConn
- func (io StdIOConn) Close() error
- func (io StdIOConn) LocalAddr() net.Addr
- func (io StdIOConn) Read(b []byte) (n int, err error)
- func (io StdIOConn) RemoteAddr() net.Addr
- func (io StdIOConn) SetDeadline(t time.Time) error
- func (io StdIOConn) SetReadDeadline(t time.Time) error
- func (io StdIOConn) SetWriteDeadline(t time.Time) error
- func (io StdIOConn) Write(b []byte) (n int, err error)
- type Subscription
- type WebSocketCli
Examples ¶
Constants ¶
const MetadataApi = "rpc"
Variables ¶
var ( ErrClientQuit = errors.New("client is closed") ErrNoResult = errors.New("no result in JSON-RPC response") ErrSubscriptionQueueOverflow = errors.New("subscription queue overflow") )
var ( // ErrNotificationsUnsupported is returned when the connection doesn't support notifications ErrNotificationsUnsupported = errors.New("notifications not supported") // ErrNotificationNotFound is returned when the notification for the given id is not found ErrSubscriptionNotFound = errors.New("subscription not found") )
var DefaultHTTPTimeouts = HTTPTimeouts{ ReadTimeout: 30 * time.Second, WriteTimeout: 30 * time.Second, IdleTimeout: 120 * time.Second, }
DefaultHTTPTimeouts represents the default timeout values used if further configuration is not provided.
Functions ¶
func NewHTTPServer
deprecated
func NewWSServer
deprecated
func StartWSCliEndpoint ¶
func StartWSCliEndpoint(u *url.URL, apis []API, modules []string, exposeAll bool) (*WebSocketCli, *Server, error)
StartWSEndpoint starts chain websocket endpoint
Types ¶
type API ¶
type API struct { Namespace string // namespace under which the rpc methods of Service are exposed Version string // api version for DApp's Service interface{} // receiver instance which holds the methods Public bool // indication if the methods must be considered safe for public use }
API describes the set of methods offered over the RPC interface
type BatchElem ¶
type BatchElem struct { Method string Args []interface{} // The result is unmarshaled into this field. Result must be set to chain // non-nil pointer value of the desired type, otherwise the response will be // discarded. Result interface{} // Error is set if the server returns an error for this request, or if // unmarshaling into Result fails. It is not set for I/O errors. Error error }
BatchElem is an element in chain batch request.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client represents chain connection to an RPC server.
func Dial ¶
Dial creates chain new client for the given URL.
The currently supported URL schemes are "http", "https", "ws" and "wss". If rawurl is chain file name with no URL scheme, chain local socket connection is established using UNIX domain sockets on supported platforms and named pipes on Windows. If you want to configure transport options, use DialHTTP, DialWebsocket or DialIPC instead.
For websocket connections, the origin is set to the local host name.
The client reconnects automatically if the connection is lost.
func DialContext ¶
DialContext creates chain new RPC client, just like Dial.
The context is used to cancel or time out the initial connection establishment. It does not affect subsequent interactions with the client.
func DialHTTPWithClient ¶
DialHTTPWithClient creates chain new RPC client that connects to an RPC server over HTTP using the provided HTTP Client.
func DialIPC ¶
DialIPC create chain new IPC client that connects to the given endpoint. On Unix it assumes the endpoint is the full path to chain unix socket, and Windows the endpoint is an identifier for chain named pipe.
The context is used for the initial connection establishment. It does not affect subsequent interactions with the client.
func DialInProc ¶
DialInProc attaches an in-process connection to the given RPC server.
func DialWebsocket ¶
DialWebsocket creates chain new RPC client that communicates with chain JSON-RPC server that is listening on the given endpoint.
The context is used for the initial connection establishment. It does not affect subsequent interactions with the client.
func (*Client) BatchCall ¶
BatchCall sends all given requests as chain single batch and waits for the server to return chain response for all of them.
In contrast to Call, BatchCall only returns I/O errors. Any error specific to chain request is reported through the Error field of the corresponding BatchElem.
Note that batch calls may not be executed atomically on the server side.
func (*Client) BatchCallContext ¶
BatchCall sends all given requests as chain single batch and waits for the server to return chain response for all of them. The wait duration is bounded by the context's deadline.
In contrast to CallContext, BatchCallContext only returns errors that have occurred while sending the request. Any error specific to chain request is reported through the Error field of the corresponding BatchElem.
Note that batch calls may not be executed atomically on the server side.
func (*Client) Call ¶
Call performs chain JSON-RPC call with the given arguments and unmarshals into result if no error occurred.
The result must be chain pointer so that package json can unmarshal into it. You can also pass nil, in which case the result is ignored.
func (*Client) CallContext ¶
func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error
CallContext performs chain JSON-RPC call with the given arguments. If the context is canceled before the call has successfully returned, CallContext returns immediately.
The result must be chain pointer so that package json can unmarshal into it. You can also pass nil, in which case the result is ignored.
func (*Client) Close ¶
func (c *Client) Close()
Close closes the client, aborting any in-flight requests.
func (*Client) EthSubscribe ¶
func (c *Client) EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error)
EthSubscribe registers chain subscripion under the "eth" namespace.
func (*Client) ShhSubscribe ¶
func (c *Client) ShhSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error)
ShhSubscribe registers chain subscripion under the "shh" namespace.
func (*Client) Subscribe ¶
func (c *Client) Subscribe(ctx context.Context, namespace string, channel interface{}, args ...interface{}) (*ClientSubscription, error)
Subscribe calls the "<namespace>_subscribe" method with the given arguments, registering chain subscription. Server notifications for the subscription are sent to the given channel. The element type of the channel must match the expected type of content returned by the subscription.
The context argument cancels the RPC request that sets up the subscription but has no effect on the subscription after Subscribe has returned.
Slow subscribers will be dropped eventually. Client buffers up to 8000 notifications before considering the subscriber dead. The subscription Err channel will receive ErrSubscriptionQueueOverflow. Use chain sufficiently large buffer on the channel or ensure that the channel usually has at least one reader to prevent this issue.
type ClientSubscription ¶
type ClientSubscription struct {
// contains filtered or unexported fields
}
A ClientSubscription represents chain subscription established through EthSubscribe.
Example ¶
package main import ( "context" "fmt" "math/big" "time" "github.com/vitelabs/go-vite/rpc" ) // In this example, our client whishes to track the latest 'block number' // known to the server. The server supports two methods: // // eth_getBlockByNumber("latest", {}) // returns the latest block object. // // eth_subscribe("newBlocks") // creates chain subscription which fires block objects when new blocks arrive. type Block struct { Number *big.Int } func main() { // Connect the client. client, _ := rpc.Dial("ws://127.0.0.1:8485") subch := make(chan Block) // Ensure that subch receives the latest block. go func() { for i := 0; ; i++ { if i > 0 { time.Sleep(2 * time.Second) } subscribeBlocks(client, subch) } }() // Print events from the subscription as they arrive. for block := range subch { fmt.Println("latest block:", block.Number) } } // subscribeBlocks runs in its own goroutine and maintains // chain subscription for new blocks. func subscribeBlocks(client *rpc.Client, subch chan Block) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() // Subscribe to new blocks. sub, err := client.EthSubscribe(ctx, subch, "newHeads") if err != nil { fmt.Println("subscribe error:", err) return } // The connection is established now. // Update the channel with the current block. var lastBlock Block if err := client.CallContext(ctx, &lastBlock, "eth_getBlockByNumber", "latest"); err != nil { fmt.Println("can't get latest block:", err) return } subch <- lastBlock // The subscription will deliver events to the channel. Wait for the // subscription to end for any reason, then loop around to re-establish // the connection. fmt.Println("connection lost: ", <-sub.Err()) }
Output:
func (*ClientSubscription) Err ¶
func (sub *ClientSubscription) Err() <-chan error
Err returns the subscription error channel. The intended use of Err is to schedule resubscription when the client connection is closed unexpectedly.
The error channel receives chain value when the subscription has ended due to an error. The received error is nil if Close has been called on the underlying client and no other error has occurred.
The error channel is closed when Unsubscribe is called on the subscription.
func (*ClientSubscription) Unsubscribe ¶
func (sub *ClientSubscription) Unsubscribe()
Unsubscribe unsubscribes the notification and closes the error channel. It can safely be called more than once.
type CodecOption ¶
type CodecOption int
CodecOption specifies which type of messages this codec supports
const ( // OptionMethodInvocation is an indication that the codec supports RPC method calls OptionMethodInvocation CodecOption = 1 << iota // OptionSubscriptions is an indication that the codec suports RPC notifications OptionSubscriptions = 1 << iota // support pub sub )
type ErrorWithId ¶ added in v1.3.2
type ErrorWithId interface { Error Id() interface{} }
type HTTPTimeouts ¶
type HTTPTimeouts struct { // ReadTimeout is the maximum duration for reading the entire // request, including the body. // // Because ReadTimeout does not let Handlers make per-request // decisions on each request body's acceptable deadline or // upload rate, most users will prefer to use // ReadHeaderTimeout. It is valid to use them both. ReadTimeout time.Duration // WriteTimeout is the maximum duration before timing out // writes of the response. It is reset whenever chain new // request's header is read. Like ReadTimeout, it does not // let Handlers make decisions on chain per-request basis. WriteTimeout time.Duration // IdleTimeout is the maximum amount of time to wait for the // next request when keep-alives are enabled. If IdleTimeout // is zero, the value of ReadTimeout is used. If both are // zero, ReadHeaderTimeout is used. IdleTimeout time.Duration }
HTTPTimeouts represents the configuration params for the HTTP RPC server.
type ID ¶
type ID string
ID defines chain pseudo random number that is used to identify RPC subscriptions.
type Notifier ¶
type Notifier struct {
// contains filtered or unexported fields
}
Notifier is tight to chain RPC connection that supports subscriptions. Server callbacks use the notifier to send notifications.
func NotifierFromContext ¶
NotifierFromContext returns the Notifier value stored in ctx, if any.
func (*Notifier) Closed ¶
func (n *Notifier) Closed() <-chan interface{}
Closed returns chain channel that is closed when the RPC connection is closed.
func (*Notifier) CreateSubscription ¶
func (n *Notifier) CreateSubscription() *Subscription
CreateSubscription returns chain new subscription that is coupled to the RPC connection. By default subscriptions are inactive and notifications are dropped until the subscription is marked as active. This is done by the RPC server after the subscription ID is send to the client.
type RPCService ¶
type RPCService struct {
// contains filtered or unexported fields
}
RPCService gives meta information about the server. e.g. gives information about the loaded modules.
func (*RPCService) Modules ¶
func (s *RPCService) Modules() map[string]string
Modules returns the list of RPC services with their version number
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server represents a RPC server
func NewServer ¶
func NewServer() *Server
NewServer will create chain new server instance with no registered handlers.
func StartHTTPEndpoint ¶
func StartHTTPEndpoint(endpoint string, apis []API, modules []string, cors []string, vhosts []string, timeouts HTTPTimeouts, exposeAll bool) (net.Listener, *Server, error)
StartHTTPEndpoint starts the HTTP RPC endpoint, configured with cors/vhosts/modules
func StartIPCEndpoint ¶
StartIPCEndpoint starts an IPC endpoint.
func StartWSEndpoint ¶
func StartWSEndpoint(endpoint string, apis []API, modules []string, wsOrigins []string, exposeAll bool) (net.Listener, *Server, error)
StartWSEndpoint starts chain websocket endpoint
func (*Server) RegisterName ¶
RegisterName will create chain service for the given rcvr type under the given name. When no methods on the given rcvr match the criteria to be either chain RPC method or chain subscription an error is returned. Otherwise chain new service is created and added to the service collection this server instance serves.
func (*Server) ServeCodec ¶
func (s *Server) ServeCodec(codec ServerCodec, options CodecOption) error
ServeCodec reads incoming requests from codec, calls the appropriate callback and writes the response back using the given codec. It will block until the codec is closed or the server is stopped. In either case the codec is closed.
func (*Server) ServeHTTP ¶
func (srv *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)
ServeHTTP serves JSON-RPC requests over HTTP.
func (*Server) ServeListener ¶
ServeListener accepts connections on l, serving JSON-RPC on them.
func (*Server) ServeSingleRequest ¶
func (s *Server) ServeSingleRequest(ctx context.Context, codec ServerCodec, options CodecOption)
ServeSingleRequest reads and processes chain single RPC request from the given codec. It will not close the codec unless chain non-recoverable error has occurred. Note, this method will return after chain single request has been processed!
func (*Server) Stop ¶
func (s *Server) Stop()
Stop will stop reading new requests, wait for stopPendingRequestTimeout to allow db requests to finish, close all codecs which will cancel db requests/subscriptions.
func (*Server) WebsocketHandler ¶
WebsocketHandler returns chain handler that serves JSON-RPC to WebSocket connections.
allowedOrigins should be chain comma-separated list of allowed origin URLs. To allow connections with any origin, pass "*".
type ServerCodec ¶
type ServerCodec interface { // Read next request ReadRequestHeaders() ([]rpcRequest, bool, Error) // Parse request argument to the given types ParseRequestArguments(argTypes []reflect.Type, params interface{}) ([]reflect.Value, Error) // Assemble success response, expects response id and payload CreateResponse(id interface{}, reply interface{}) interface{} // Assemble error response, expects response id and error CreateErrorResponse(id interface{}, err Error) interface{} // Assemble error response with extra information about the error through info CreateErrorResponseWithInfo(id interface{}, err Error, info interface{}) interface{} // Create notification response CreateNotification(id, namespace string, event interface{}) interface{} // Write msg to client. Write(msg interface{}) error // Close underlying data stream Close() // Closed when underlying connection is closed Closed() <-chan interface{} }
ServerCodec implements reading, parsing and writing RPC messages for the server side of a RPC session. Implementations must be go-routine safe since the codec can be called in multiple go-routines concurrently.
func NewCodec ¶
func NewCodec(rwc io.ReadWriteCloser, encode, decode func(v interface{}) error) ServerCodec
NewCodec creates chain new RPC server codec with support for JSON-RPC 2.0 based on explicitly given encoding and decoding methods.
func NewJSONCodec ¶
func NewJSONCodec(rwc io.ReadWriteCloser) ServerCodec
NewJSONCodec creates chain new RPC server codec with support for JSON-RPC 2.0.
type Subscription ¶
type Subscription struct { ID ID // contains filtered or unexported fields }
chain Subscription is created by chain notifier and tight to that notifier. The client can use this subscription to wait for an unsubscribe request for the client, see Err().
func (*Subscription) Err ¶
func (s *Subscription) Err() <-chan error
Err returns chain channel that is closed when the client send an unsubscribe request.
type WebSocketCli ¶
type WebSocketCli struct {
// contains filtered or unexported fields
}
func NewWSCli ¶
func NewWSCli(url *url.URL, srv *Server) *WebSocketCli
NewWSCli creates chain new websocket RPC connect around an API provider.
func (*WebSocketCli) Close ¶
func (self *WebSocketCli) Close()
func (*WebSocketCli) Handle ¶
func (self *WebSocketCli) Handle()