lotus/lib/jsonrpc/client.go

360 lines
7.9 KiB
Go
Raw Normal View History

2019-07-03 17:39:07 +00:00
package jsonrpc
import (
2019-09-27 11:37:44 +00:00
"container/list"
"context"
"encoding/base64"
"encoding/json"
"fmt"
2019-07-23 18:49:09 +00:00
"net/http"
"reflect"
2019-09-27 11:37:44 +00:00
"sync"
2019-07-03 14:32:31 +00:00
"sync/atomic"
2019-07-08 19:07:16 +00:00
2019-07-12 15:29:41 +00:00
"github.com/gorilla/websocket"
2019-07-08 19:07:16 +00:00
logging "github.com/ipfs/go-log"
"go.opencensus.io/trace"
"go.opencensus.io/trace/propagation"
2019-07-18 10:46:32 +00:00
"golang.org/x/xerrors"
)
2019-07-08 19:07:16 +00:00
var log = logging.Logger("rpc")
var (
errorType = reflect.TypeOf(new(error)).Elem()
contextType = reflect.TypeOf(new(context.Context)).Elem()
)
2019-07-02 19:08:30 +00:00
// ErrClient is an error which occurred on the client side the library
type ErrClient struct {
err error
}
func (e *ErrClient) Error() string {
return fmt.Sprintf("RPC client error: %s", e.err)
}
2019-07-02 17:45:03 +00:00
// Unwrap unwraps the actual error
func (e *ErrClient) Unwrap(err error) error {
return e.err
}
type clientResponse struct {
2019-07-22 20:28:40 +00:00
Jsonrpc string `json:"jsonrpc"`
Result json.RawMessage `json:"result"`
ID int64 `json:"id"`
Error *respError `json:"error,omitempty"`
}
type makeChanSink func() (context.Context, func([]byte, bool))
2019-07-12 15:29:41 +00:00
type clientRequest struct {
req request
ready chan clientResponse
2019-07-22 18:13:41 +00:00
2019-07-22 23:58:43 +00:00
// retCh provides a context and sink for handling incoming channel messages
retCh makeChanSink
2019-07-12 15:29:41 +00:00
}
// ClientCloser is used to close Client from further use
type ClientCloser func()
2019-09-17 14:34:22 +00:00
// NewClient creates new jsonrpc 2.0 client
2019-07-02 17:45:03 +00:00
//
// handler must be pointer to a struct with function fields
2019-07-23 19:38:10 +00:00
// Returned value closes the client connection
2019-07-02 17:45:03 +00:00
// TODO: Example
2019-07-23 18:49:09 +00:00
func NewClient(addr string, namespace string, handler interface{}, requestHeader http.Header) (ClientCloser, error) {
return NewMergeClient(addr, namespace, []interface{}{handler}, requestHeader)
}
type client struct {
namespace string
requests chan clientRequest
2019-07-25 12:54:19 +00:00
idCtr int64
}
// NewMergeClient is like NewClient, but allows to specify multiple structs
// to be filled in the same namespace, using one connection
func NewMergeClient(addr string, namespace string, outs []interface{}, requestHeader http.Header) (ClientCloser, error) {
2019-07-23 18:49:09 +00:00
conn, _, err := websocket.DefaultDialer.Dial(addr, requestHeader)
2019-07-12 15:29:41 +00:00
if err != nil {
return nil, err
}
c := client{
namespace: namespace,
}
2019-07-12 15:29:41 +00:00
stop := make(chan struct{})
2019-10-08 08:29:37 +00:00
exiting := make(chan struct{})
c.requests = make(chan clientRequest)
2019-07-12 15:29:41 +00:00
2019-07-12 17:12:38 +00:00
handlers := map[string]rpcHandler{}
go (&wsConn{
conn: conn,
handler: handlers,
requests: c.requests,
stop: stop,
2019-10-08 08:29:37 +00:00
exiting: exiting,
}).handleWsConn(context.TODO())
2019-07-12 15:29:41 +00:00
for _, handler := range outs {
htyp := reflect.TypeOf(handler)
if htyp.Kind() != reflect.Ptr {
return nil, xerrors.New("expected handler to be a pointer")
}
typ := htyp.Elem()
if typ.Kind() != reflect.Struct {
return nil, xerrors.New("handler should be a struct")
}
val := reflect.ValueOf(handler)
for i := 0; i < typ.NumField(); i++ {
fn, err := c.makeRpcFunc(typ.Field(i))
if err != nil {
return nil, err
}
val.Elem().Field(i).Set(fn)
}
}
return func() {
close(stop)
2019-10-08 08:29:37 +00:00
<-exiting
}, nil
}
func (c *client) makeOutChan(ctx context.Context, ftyp reflect.Type, valOut int) (func() reflect.Value, makeChanSink) {
retVal := reflect.Zero(ftyp.Out(valOut))
chCtor := func() (context.Context, func([]byte, bool)) {
// unpack chan type to make sure it's reflect.BothDir
ctyp := reflect.ChanOf(reflect.BothDir, ftyp.Out(valOut).Elem())
ch := reflect.MakeChan(ctyp, 0) // todo: buffer?
retVal = ch.Convert(ftyp.Out(valOut))
2019-09-27 11:37:44 +00:00
buf := (&list.List{}).Init()
var bufLk sync.Mutex
return ctx, func(result []byte, ok bool) {
if !ok {
// remote channel closed, close ours too
ch.Close()
return
}
val := reflect.New(ftyp.Out(valOut).Elem())
if err := json.Unmarshal(result, val.Interface()); err != nil {
log.Errorf("error unmarshaling chan response: %s", err)
return
}
2019-09-27 11:37:44 +00:00
bufLk.Lock()
if ctx.Err() != nil {
log.Errorf("got rpc message with cancelled context: %s", ctx.Err())
bufLk.Unlock()
return
}
buf.PushBack(val)
if buf.Len() > 1 {
log.Warnw("rpc output message buffer", "n", buf.Len())
2019-09-27 11:37:44 +00:00
bufLk.Unlock()
return
}
go func() {
for buf.Len() > 0 {
front := buf.Front()
bufLk.Unlock()
ch.Send(front.Value.(reflect.Value).Elem()) // todo: select on ctx is probably a good idea
bufLk.Lock()
buf.Remove(front)
}
bufLk.Unlock()
}()
2019-07-25 12:54:19 +00:00
}
}
return func() reflect.Value { return retVal }, chCtor
}
func (c *client) sendRequest(ctx context.Context, req request, chCtor makeChanSink) clientResponse {
rchan := make(chan clientResponse, 1)
c.requests <- clientRequest{
req: req,
ready: rchan,
retCh: chCtor,
}
var ctxDone <-chan struct{}
var resp clientResponse
if ctx != nil {
ctxDone = ctx.Done()
}
// wait for response, handle context cancellation
loop:
for {
select {
case resp = <-rchan:
break loop
case <-ctxDone: // send cancel request
ctxDone = nil
c.requests <- clientRequest{
2019-09-27 11:37:44 +00:00
req: request{
Jsonrpc: "2.0",
Method: wsCancel,
Params: []param{{v: reflect.ValueOf(*req.ID)}},
},
}
}
}
return resp
}
type rpcFunc struct {
client *client
ftyp reflect.Type
name string
2019-07-25 12:54:19 +00:00
nout int
valOut int
errOut int
hasCtx int
2019-07-25 12:54:19 +00:00
retCh bool
}
func (fn *rpcFunc) processResponse(resp clientResponse, rval reflect.Value) []reflect.Value {
out := make([]reflect.Value, fn.nout)
if fn.valOut != -1 {
out[fn.valOut] = rval
}
if fn.errOut != -1 {
out[fn.errOut] = reflect.New(errorType).Elem()
if resp.Error != nil {
out[fn.errOut].Set(reflect.ValueOf(resp.Error))
}
}
2019-07-03 14:32:31 +00:00
return out
}
func (fn *rpcFunc) processError(err error) []reflect.Value {
out := make([]reflect.Value, fn.nout)
if fn.valOut != -1 {
out[fn.valOut] = reflect.New(fn.ftyp.Out(fn.valOut)).Elem()
}
if fn.errOut != -1 {
out[fn.errOut] = reflect.New(errorType).Elem()
out[fn.errOut].Set(reflect.ValueOf(&ErrClient{err}))
}
return out
}
func (fn *rpcFunc) handleRpcCall(args []reflect.Value) (results []reflect.Value) {
id := atomic.AddInt64(&fn.client.idCtr, 1)
params := make([]param, len(args)-fn.hasCtx)
for i, arg := range args[fn.hasCtx:] {
params[i] = param{
v: arg,
}
}
var ctx context.Context
var span *trace.Span
if fn.hasCtx == 1 {
ctx = args[0].Interface().(context.Context)
ctx, span = trace.StartSpan(ctx, "api.call")
defer span.End()
}
retVal := func() reflect.Value { return reflect.Value{} }
// if the function returns a channel, we need to provide a sink for the
// messages
var chCtor makeChanSink
if fn.retCh {
retVal, chCtor = fn.client.makeOutChan(ctx, fn.ftyp, fn.valOut)
}
req := request{
Jsonrpc: "2.0",
ID: &id,
Method: fn.client.namespace + "." + fn.name,
Params: params,
}
if span != nil {
span.AddAttributes(trace.StringAttribute("method", req.Method))
eSC := base64.StdEncoding.EncodeToString(
propagation.Binary(span.SpanContext()))
req.Meta = map[string]string{
"SpanContext": eSC,
}
}
resp := fn.client.sendRequest(ctx, req, chCtor)
if resp.ID != *req.ID {
return fn.processError(xerrors.New("request and response id didn't match"))
}
if fn.valOut != -1 && !fn.retCh {
val := reflect.New(fn.ftyp.Out(fn.valOut))
if resp.Result != nil {
log.Debugw("rpc result", "type", fn.ftyp.Out(fn.valOut))
if err := json.Unmarshal(resp.Result, val.Interface()); err != nil {
2019-08-26 10:04:57 +00:00
log.Warnw("unmarshaling failed", "message", string(resp.Result))
return fn.processError(xerrors.Errorf("unmarshaling result: %w", err))
}
}
retVal = func() reflect.Value { return val.Elem() }
}
return fn.processResponse(resp, retVal())
}
func (c *client) makeRpcFunc(f reflect.StructField) (reflect.Value, error) {
ftyp := f.Type
if ftyp.Kind() != reflect.Func {
return reflect.Value{}, xerrors.New("handler field not a func")
}
fun := &rpcFunc{
client: c,
2019-07-25 12:54:19 +00:00
ftyp: ftyp,
name: f.Name,
}
fun.valOut, fun.errOut, fun.nout = processFuncOut(ftyp)
if ftyp.NumIn() > 0 && ftyp.In(0) == contextType {
fun.hasCtx = 1
}
fun.retCh = fun.valOut != -1 && ftyp.Out(fun.valOut).Kind() == reflect.Chan
return reflect.MakeFunc(ftyp, fun.handleRpcCall), nil
}