package rpc import ( "bytes" "encoding/json" "errors" "fmt" "github.com/gorilla/websocket" "github.com/tendermint/tendermint/binary" "github.com/tendermint/tendermint/events" "io/ioutil" "net/http" "reflect" "sync/atomic" "time" ) func RegisterRPCFuncs(mux *http.ServeMux, funcMap map[string]*RPCFunc) { // HTTP endpoints for funcName, rpcFunc := range funcMap { mux.HandleFunc("/"+funcName, makeHTTPHandler(rpcFunc)) } // JSONRPC endpoints mux.HandleFunc("/", makeJSONRPCHandler(funcMap)) } func RegisterEventsHandler(mux *http.ServeMux, evsw *events.EventSwitch) { // websocket endpoint wm := NewWebsocketManager(evsw) mux.HandleFunc("/events", wm.websocketHandler) // websocket.Handler(w.eventsHandler)) } //------------------------------------- // function introspection // holds all type information for each function type RPCFunc struct { f reflect.Value // underlying rpc function args []reflect.Type // type of each function arg returns []reflect.Type // type of each return arg argNames []string // name of each argument } // wraps a function for quicker introspection func NewRPCFunc(f interface{}, args []string) *RPCFunc { return &RPCFunc{ f: reflect.ValueOf(f), args: funcArgTypes(f), returns: funcReturnTypes(f), argNames: args, } } // return a function's argument types func funcArgTypes(f interface{}) []reflect.Type { t := reflect.TypeOf(f) n := t.NumIn() types := make([]reflect.Type, n) for i := 0; i < n; i++ { types[i] = t.In(i) } return types } // return a function's return types func funcReturnTypes(f interface{}) []reflect.Type { t := reflect.TypeOf(f) n := t.NumOut() types := make([]reflect.Type, n) for i := 0; i < n; i++ { types[i] = t.Out(i) } return types } // function introspection //----------------------------------------------------------------------------- // rpc.json // jsonrpc calls grab the given method's function info and runs reflect.Call func makeJSONRPCHandler(funcMap map[string]*RPCFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if len(r.URL.Path) > 1 { WriteRPCResponse(w, NewRPCResponse(nil, fmt.Sprintf("Invalid JSONRPC endpoint %s", r.URL.Path))) return } b, _ := ioutil.ReadAll(r.Body) var request RPCRequest err := json.Unmarshal(b, &request) if err != nil { WriteRPCResponse(w, NewRPCResponse(nil, err.Error())) return } rpcFunc := funcMap[request.Method] if rpcFunc == nil { WriteRPCResponse(w, NewRPCResponse(nil, "RPC method unknown: "+request.Method)) return } args, err := jsonParamsToArgs(rpcFunc, request.Params) if err != nil { WriteRPCResponse(w, NewRPCResponse(nil, err.Error())) return } returns := rpcFunc.f.Call(args) log.Debug("HTTPJSONRPC", "method", request.Method, "args", args, "returns", returns) response, err := unreflectResponse(returns) if err != nil { WriteRPCResponse(w, NewRPCResponse(nil, err.Error())) return } WriteRPCResponse(w, NewRPCResponse(response, "")) } } // covert a list of interfaces to properly typed values func jsonParamsToArgs(rpcFunc *RPCFunc, params []interface{}) ([]reflect.Value, error) { if len(rpcFunc.argNames) != len(params) { return nil, errors.New(fmt.Sprintf("Expected %v parameters (%v), got %v (%v)", len(rpcFunc.argNames), rpcFunc.argNames, len(params), params)) } values := make([]reflect.Value, len(params)) for i, p := range params { ty := rpcFunc.args[i] v, err := _jsonObjectToArg(ty, p) if err != nil { return nil, err } values[i] = v } return values, nil } func _jsonObjectToArg(ty reflect.Type, object interface{}) (reflect.Value, error) { var err error v := reflect.New(ty) binary.ReadJSONObject(v.Interface(), object, &err) if err != nil { return v, err } v = v.Elem() return v, nil } // rpc.json //----------------------------------------------------------------------------- // rpc.http // convert from a function name to the http handler func makeHTTPHandler(rpcFunc *RPCFunc) func(http.ResponseWriter, *http.Request) { return func(w http.ResponseWriter, r *http.Request) { args, err := httpParamsToArgs(rpcFunc, r) if err != nil { WriteRPCResponse(w, NewRPCResponse(nil, err.Error())) return } returns := rpcFunc.f.Call(args) log.Debug("HTTPRestRPC", "method", r.URL.Path, "args", args, "returns", returns) response, err := unreflectResponse(returns) if err != nil { WriteRPCResponse(w, NewRPCResponse(nil, err.Error())) return } WriteRPCResponse(w, NewRPCResponse(response, "")) } } // Covert an http query to a list of properly typed values. // To be properly decoded the arg must be a concrete type from tendermint (if its an interface). func httpParamsToArgs(rpcFunc *RPCFunc, r *http.Request) ([]reflect.Value, error) { argTypes := rpcFunc.args argNames := rpcFunc.argNames var err error values := make([]reflect.Value, len(argNames)) for i, name := range argNames { ty := argTypes[i] arg := GetParam(r, name) values[i], err = _jsonStringToArg(ty, arg) if err != nil { return nil, err } } return values, nil } func _jsonStringToArg(ty reflect.Type, arg string) (reflect.Value, error) { var err error v := reflect.New(ty) binary.ReadJSON(v.Interface(), []byte(arg), &err) if err != nil { return v, err } v = v.Elem() return v, nil } // rpc.http //----------------------------------------------------------------------------- // rpc.websocket const ( WSConnectionReaperSeconds = 5 MaxFailedSends = 10 WriteChanBufferSize = 10 ) // for requests coming in type WSRequest struct { Type string // subscribe or unsubscribe Event string } // for responses going out type WSResponse struct { Event string Data interface{} Error string } // a single websocket connection // contains listener id, underlying ws connection, // and the event switch for subscribing to events type WSConnection struct { id string wsConn *websocket.Conn writeChan chan WSResponse quitChan chan struct{} failedSends uint started uint32 stopped uint32 evsw *events.EventSwitch } // new websocket connection wrapper func NewWSConnection(wsConn *websocket.Conn) *WSConnection { return &WSConnection{ id: wsConn.RemoteAddr().String(), wsConn: wsConn, writeChan: make(chan WSResponse, WriteChanBufferSize), // buffered. we keep track when its full quitChan: make(chan struct{}), } } // start the connection and hand her the event switch func (con *WSConnection) Start(evsw *events.EventSwitch) { if atomic.CompareAndSwapUint32(&con.started, 0, 1) { con.evsw = evsw // read subscriptions/unsubscriptions to events go con.read() // write responses con.write() } } // close the connection func (con *WSConnection) Stop() { if atomic.CompareAndSwapUint32(&con.stopped, 0, 1) { close(con.quitChan) // the write loop closes the websocket connection // when it exits its loop, and the read loop // closes the writeChan } } // attempt to write response to writeChan and record failures func (con *WSConnection) safeWrite(resp WSResponse) { select { case con.writeChan <- resp: // yay con.failedSends = 0 default: // channel is full // if this happens too many times in a row, // close connection con.failedSends += 1 } } // read from the socket and subscribe to or unsubscribe from events func (con *WSConnection) read() { defer close(con.writeChan) reaper := time.Tick(time.Second * WSConnectionReaperSeconds) for { select { case <-reaper: if con.failedSends > MaxFailedSends { // sending has failed too many times. // kill the connection con.Stop() return } default: var in []byte _, in, err := con.wsConn.ReadMessage() if err != nil { // an error reading the connection, // kill the connection con.Stop() return } var req WSRequest err = json.Unmarshal(in, &req) if err != nil { errStr := fmt.Sprintf("Error unmarshaling data: %s", err.Error()) con.safeWrite(WSResponse{Error: errStr}) continue } switch req.Type { case "subscribe": log.Info("New event subscription", "con id", con.id, "event", req.Event) con.evsw.AddListenerForEvent(con.id, req.Event, func(msg interface{}) { resp := WSResponse{ Event: req.Event, Data: msg, } con.safeWrite(resp) }) case "unsubscribe": if req.Event != "" { con.evsw.RemoveListenerForEvent(req.Event, con.id) } else { con.evsw.RemoveListener(con.id) } default: con.safeWrite(WSResponse{Error: "Unknown request type: " + req.Type}) } } } } // receives on a write channel and writes out on the socket func (con *WSConnection) write() { defer con.wsConn.Close() n, err := new(int64), new(error) for { select { case msg := <-con.writeChan: buf := new(bytes.Buffer) binary.WriteJSON(msg, buf, n, err) if *err != nil { log.Error("Failed to marshal WSResponse to JSON", "error", err) } else { if err := con.wsConn.WriteMessage(websocket.TextMessage, buf.Bytes()); err != nil { log.Error("Failed to write response on websocket", "error", err) con.Stop() return } } case <-con.quitChan: return } } } // main manager for all websocket connections // holds the event switch type WebsocketManager struct { websocket.Upgrader evsw *events.EventSwitch } func NewWebsocketManager(evsw *events.EventSwitch) *WebsocketManager { return &WebsocketManager{ evsw: evsw, Upgrader: websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, CheckOrigin: func(r *http.Request) bool { // TODO return true }, }, } } func (wm *WebsocketManager) websocketHandler(w http.ResponseWriter, r *http.Request) { wsConn, err := wm.Upgrade(w, r, nil) if err != nil { // TODO - return http error log.Error("Failed to upgrade to websocket connection", "error", err) return } // register connection con := NewWSConnection(wsConn) log.Info("New websocket connection", "origin", con.id) con.Start(wm.evsw) } // rpc.websocket //----------------------------------------------------------------------------- // returns is Response struct and error. If error is not nil, return it func unreflectResponse(returns []reflect.Value) (interface{}, error) { errV := returns[1] if errV.Interface() != nil { return nil, fmt.Errorf("%v", errV.Interface()) } return returns[0].Interface(), nil }