Browse Source

Fix rpc/lib/...

pull/1347/head
Jae Kwon 7 years ago
parent
commit
3037b5b7ca
10 changed files with 132 additions and 72 deletions
  1. +5
    -1
      rpc/lib/client/args_test.go
  2. +27
    -14
      rpc/lib/client/http_client.go
  3. +9
    -2
      rpc/lib/client/ws_client.go
  4. +8
    -4
      rpc/lib/rpc_test.go
  5. +48
    -41
      rpc/lib/server/handlers.go
  6. +3
    -1
      rpc/lib/server/handlers_test.go
  7. +3
    -1
      rpc/lib/server/parse_test.go
  8. +3
    -1
      rpc/lib/test/main.go
  9. +23
    -6
      rpc/lib/types/types.go
  10. +3
    -1
      rpc/lib/types/types_test.go

+ 5
- 1
rpc/lib/client/args_test.go View File

@ -5,6 +5,8 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/tendermint/go-amino"
) )
type Tx []byte type Tx []byte
@ -27,9 +29,11 @@ func TestArgToJSON(t *testing.T) {
{Foo{7, "hello"}, `{"Bar":7,"Baz":"hello"}`}, {Foo{7, "hello"}, `{"Bar":7,"Baz":"hello"}`},
} }
cdc := amino.NewCodec()
for i, tc := range cases { for i, tc := range cases {
args := map[string]interface{}{"data": tc.input} args := map[string]interface{}{"data": tc.input}
err := argsToJson(args)
err := argsToJSON(cdc, args)
require.Nil(err, "%d: %+v", i, err) require.Nil(err, "%d: %+v", i, err)
require.Equal(1, len(args), "%d", i) require.Equal(1, len(args), "%d", i)
data, ok := args["data"].(string) data, ok := args["data"].(string)


+ 27
- 14
rpc/lib/client/http_client.go View File

@ -12,6 +12,7 @@ import (
"strings" "strings"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/tendermint/go-amino"
types "github.com/tendermint/tendermint/rpc/lib/types" types "github.com/tendermint/tendermint/rpc/lib/types"
) )
@ -19,6 +20,7 @@ import (
// HTTPClient is a common interface for JSONRPCClient and URIClient. // HTTPClient is a common interface for JSONRPCClient and URIClient.
type HTTPClient interface { type HTTPClient interface {
Call(method string, params map[string]interface{}, result interface{}) (interface{}, error) Call(method string, params map[string]interface{}, result interface{}) (interface{}, error)
Codec() *amino.Codec
} }
// TODO: Deprecate support for IP:PORT or /path/to/socket // TODO: Deprecate support for IP:PORT or /path/to/socket
@ -66,6 +68,7 @@ func makeHTTPClient(remoteAddr string) (string, *http.Client) {
type JSONRPCClient struct { type JSONRPCClient struct {
address string address string
client *http.Client client *http.Client
cdc *amino.Codec
} }
// NewJSONRPCClient returns a JSONRPCClient pointed at the given address. // NewJSONRPCClient returns a JSONRPCClient pointed at the given address.
@ -74,11 +77,12 @@ func NewJSONRPCClient(remote string) *JSONRPCClient {
return &JSONRPCClient{ return &JSONRPCClient{
address: address, address: address,
client: client, client: client,
cdc: amino.NewCodec(),
} }
} }
func (c *JSONRPCClient) Call(method string, params map[string]interface{}, result interface{}) (interface{}, error) { func (c *JSONRPCClient) Call(method string, params map[string]interface{}, result interface{}) (interface{}, error) {
request, err := types.MapToRequest("jsonrpc-client", method, params)
request, err := types.MapToRequest(c.cdc, "jsonrpc-client", method, params)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -100,7 +104,11 @@ func (c *JSONRPCClient) Call(method string, params map[string]interface{}, resul
return nil, err return nil, err
} }
// log.Info(Fmt("RPC response: %v", string(responseBytes))) // log.Info(Fmt("RPC response: %v", string(responseBytes)))
return unmarshalResponseBytes(responseBytes, result)
return unmarshalResponseBytes(c.cdc, responseBytes, result)
}
func (c *JSONRPCClient) Codec() *amino.Codec {
return c.cdc
} }
//------------------------------------------------------------- //-------------------------------------------------------------
@ -109,6 +117,7 @@ func (c *JSONRPCClient) Call(method string, params map[string]interface{}, resul
type URIClient struct { type URIClient struct {
address string address string
client *http.Client client *http.Client
cdc *amino.Codec
} }
func NewURIClient(remote string) *URIClient { func NewURIClient(remote string) *URIClient {
@ -116,11 +125,12 @@ func NewURIClient(remote string) *URIClient {
return &URIClient{ return &URIClient{
address: address, address: address,
client: client, client: client,
cdc: amino.NewCodec(),
} }
} }
func (c *URIClient) Call(method string, params map[string]interface{}, result interface{}) (interface{}, error) { func (c *URIClient) Call(method string, params map[string]interface{}, result interface{}) (interface{}, error) {
values, err := argsToURLValues(params)
values, err := argsToURLValues(c.cdc, params)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -135,15 +145,18 @@ func (c *URIClient) Call(method string, params map[string]interface{}, result in
if err != nil { if err != nil {
return nil, err return nil, err
} }
return unmarshalResponseBytes(responseBytes, result)
return unmarshalResponseBytes(c.cdc, responseBytes, result)
}
func (c *URIClient) Codec() *amino.Codec {
return c.cdc
} }
//------------------------------------------------ //------------------------------------------------
func unmarshalResponseBytes(responseBytes []byte, result interface{}) (interface{}, error) {
// read response
// if rpc/core/types is imported, the result will unmarshal
// into the correct type
func unmarshalResponseBytes(cdc *amino.Codec, responseBytes []byte, result interface{}) (interface{}, error) {
// Read response. If rpc/core/types is imported, the result will unmarshal
// into the correct type.
// log.Notice("response", "response", string(responseBytes)) // log.Notice("response", "response", string(responseBytes))
var err error var err error
response := &types.RPCResponse{} response := &types.RPCResponse{}
@ -154,20 +167,20 @@ func unmarshalResponseBytes(responseBytes []byte, result interface{}) (interface
if response.Error != nil { if response.Error != nil {
return nil, errors.Errorf("Response error: %v", response.Error) return nil, errors.Errorf("Response error: %v", response.Error)
} }
// unmarshal the RawMessage into the result
err = json.Unmarshal(response.Result, result)
// Unmarshal the RawMessage into the result.
err = cdc.UnmarshalJSON(response.Result, result)
if err != nil { if err != nil {
return nil, errors.Errorf("Error unmarshalling rpc response result: %v", err) return nil, errors.Errorf("Error unmarshalling rpc response result: %v", err)
} }
return result, nil return result, nil
} }
func argsToURLValues(args map[string]interface{}) (url.Values, error) {
func argsToURLValues(cdc *amino.Codec, args map[string]interface{}) (url.Values, error) {
values := make(url.Values) values := make(url.Values)
if len(args) == 0 { if len(args) == 0 {
return values, nil return values, nil
} }
err := argsToJson(args)
err := argsToJSON(cdc, args)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -177,7 +190,7 @@ func argsToURLValues(args map[string]interface{}) (url.Values, error) {
return values, nil return values, nil
} }
func argsToJson(args map[string]interface{}) error {
func argsToJSON(cdc *amino.Codec, args map[string]interface{}) error {
for k, v := range args { for k, v := range args {
rt := reflect.TypeOf(v) rt := reflect.TypeOf(v)
isByteSlice := rt.Kind() == reflect.Slice && rt.Elem().Kind() == reflect.Uint8 isByteSlice := rt.Kind() == reflect.Slice && rt.Elem().Kind() == reflect.Uint8
@ -187,7 +200,7 @@ func argsToJson(args map[string]interface{}) error {
continue continue
} }
data, err := json.Marshal(v)
data, err := cdc.MarshalJSON(v)
if err != nil { if err != nil {
return err return err
} }


+ 9
- 2
rpc/lib/client/ws_client.go View File

@ -14,6 +14,7 @@ import (
"github.com/pkg/errors" "github.com/pkg/errors"
metrics "github.com/rcrowley/go-metrics" metrics "github.com/rcrowley/go-metrics"
"github.com/tendermint/go-amino"
types "github.com/tendermint/tendermint/rpc/lib/types" types "github.com/tendermint/tendermint/rpc/lib/types"
cmn "github.com/tendermint/tmlibs/common" cmn "github.com/tendermint/tmlibs/common"
) )
@ -31,6 +32,7 @@ type WSClient struct {
cmn.BaseService cmn.BaseService
conn *websocket.Conn conn *websocket.Conn
cdc *amino.Codec
Address string // IP:PORT or /path/to/socket Address string // IP:PORT or /path/to/socket
Endpoint string // /websocket/url/endpoint Endpoint string // /websocket/url/endpoint
@ -77,6 +79,7 @@ type WSClient struct {
func NewWSClient(remoteAddr, endpoint string, options ...func(*WSClient)) *WSClient { func NewWSClient(remoteAddr, endpoint string, options ...func(*WSClient)) *WSClient {
addr, dialer := makeHTTPDialer(remoteAddr) addr, dialer := makeHTTPDialer(remoteAddr)
c := &WSClient{ c := &WSClient{
cdc: amino.NewCodec(),
Address: addr, Address: addr,
Dialer: dialer, Dialer: dialer,
Endpoint: endpoint, Endpoint: endpoint,
@ -206,7 +209,7 @@ func (c *WSClient) Send(ctx context.Context, request types.RPCRequest) error {
// Call the given method. See Send description. // Call the given method. See Send description.
func (c *WSClient) Call(ctx context.Context, method string, params map[string]interface{}) error { func (c *WSClient) Call(ctx context.Context, method string, params map[string]interface{}) error {
request, err := types.MapToRequest("ws-client", method, params)
request, err := types.MapToRequest(c.cdc, "ws-client", method, params)
if err != nil { if err != nil {
return err return err
} }
@ -216,13 +219,17 @@ func (c *WSClient) Call(ctx context.Context, method string, params map[string]in
// CallWithArrayParams the given method with params in a form of array. See // CallWithArrayParams the given method with params in a form of array. See
// Send description. // Send description.
func (c *WSClient) CallWithArrayParams(ctx context.Context, method string, params []interface{}) error { func (c *WSClient) CallWithArrayParams(ctx context.Context, method string, params []interface{}) error {
request, err := types.ArrayToRequest("ws-client", method, params)
request, err := types.ArrayToRequest(c.cdc, "ws-client", method, params)
if err != nil { if err != nil {
return err return err
} }
return c.Send(ctx, request) return c.Send(ctx, request)
} }
func (c *WSClient) Codec() *amino.Codec {
return c.cdc
}
/////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////
// Private methods // Private methods


+ 8
- 4
rpc/lib/rpc_test.go View File

@ -17,6 +17,7 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/tendermint/go-amino"
cmn "github.com/tendermint/tmlibs/common" cmn "github.com/tendermint/tmlibs/common"
"github.com/tendermint/tmlibs/log" "github.com/tendermint/tmlibs/log"
@ -60,6 +61,9 @@ var Routes = map[string]*server.RPCFunc{
"echo_int": server.NewRPCFunc(EchoIntResult, "arg"), "echo_int": server.NewRPCFunc(EchoIntResult, "arg"),
} }
// Amino codec required to encode/decode everything above.
var RoutesCdc = amino.NewCodec()
func EchoResult(v string) (*ResultEcho, error) { func EchoResult(v string) (*ResultEcho, error) {
return &ResultEcho{v}, nil return &ResultEcho{v}, nil
} }
@ -114,8 +118,8 @@ func setup() {
tcpLogger := logger.With("socket", "tcp") tcpLogger := logger.With("socket", "tcp")
mux := http.NewServeMux() mux := http.NewServeMux()
server.RegisterRPCFuncs(mux, Routes, tcpLogger)
wm := server.NewWebsocketManager(Routes, server.ReadWait(5*time.Second), server.PingPeriod(1*time.Second))
server.RegisterRPCFuncs(mux, Routes, RoutesCdc, tcpLogger)
wm := server.NewWebsocketManager(Routes, RoutesCdc, server.ReadWait(5*time.Second), server.PingPeriod(1*time.Second))
wm.SetLogger(tcpLogger) wm.SetLogger(tcpLogger)
mux.HandleFunc(websocketEndpoint, wm.WebsocketHandler) mux.HandleFunc(websocketEndpoint, wm.WebsocketHandler)
go func() { go func() {
@ -127,8 +131,8 @@ func setup() {
unixLogger := logger.With("socket", "unix") unixLogger := logger.With("socket", "unix")
mux2 := http.NewServeMux() mux2 := http.NewServeMux()
server.RegisterRPCFuncs(mux2, Routes, unixLogger)
wm = server.NewWebsocketManager(Routes)
server.RegisterRPCFuncs(mux2, Routes, RoutesCdc, unixLogger)
wm = server.NewWebsocketManager(Routes, RoutesCdc)
wm.SetLogger(unixLogger) wm.SetLogger(unixLogger)
mux2.HandleFunc(websocketEndpoint, wm.WebsocketHandler) mux2.HandleFunc(websocketEndpoint, wm.WebsocketHandler)
go func() { go func() {


+ 48
- 41
rpc/lib/server/handlers.go View File

@ -17,6 +17,7 @@ import (
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/tendermint/go-amino"
types "github.com/tendermint/tendermint/rpc/lib/types" types "github.com/tendermint/tendermint/rpc/lib/types"
cmn "github.com/tendermint/tmlibs/common" cmn "github.com/tendermint/tmlibs/common"
"github.com/tendermint/tmlibs/log" "github.com/tendermint/tmlibs/log"
@ -24,14 +25,14 @@ import (
// RegisterRPCFuncs adds a route for each function in the funcMap, as well as general jsonrpc and websocket handlers for all functions. // RegisterRPCFuncs adds a route for each function in the funcMap, as well as general jsonrpc and websocket handlers for all functions.
// "result" is the interface on which the result objects are registered, and is popualted with every RPCResponse // "result" is the interface on which the result objects are registered, and is popualted with every RPCResponse
func RegisterRPCFuncs(mux *http.ServeMux, funcMap map[string]*RPCFunc, logger log.Logger) {
func RegisterRPCFuncs(mux *http.ServeMux, funcMap map[string]*RPCFunc, cdc *amino.Codec, logger log.Logger) {
// HTTP endpoints // HTTP endpoints
for funcName, rpcFunc := range funcMap { for funcName, rpcFunc := range funcMap {
mux.HandleFunc("/"+funcName, makeHTTPHandler(rpcFunc, logger))
mux.HandleFunc("/"+funcName, makeHTTPHandler(rpcFunc, cdc, logger))
} }
// JSONRPC endpoints // JSONRPC endpoints
mux.HandleFunc("/", makeJSONRPCHandler(funcMap, logger))
mux.HandleFunc("/", makeJSONRPCHandler(funcMap, cdc, logger))
} }
//------------------------------------- //-------------------------------------
@ -98,7 +99,7 @@ func funcReturnTypes(f interface{}) []reflect.Type {
// rpc.json // rpc.json
// jsonrpc calls grab the given method's function info and runs reflect.Call // jsonrpc calls grab the given method's function info and runs reflect.Call
func makeJSONRPCHandler(funcMap map[string]*RPCFunc, logger log.Logger) http.HandlerFunc {
func makeJSONRPCHandler(funcMap map[string]*RPCFunc, cdc *amino.Codec, logger log.Logger) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
b, err := ioutil.ReadAll(r.Body) b, err := ioutil.ReadAll(r.Body)
if err != nil { if err != nil {
@ -135,7 +136,7 @@ func makeJSONRPCHandler(funcMap map[string]*RPCFunc, logger log.Logger) http.Han
} }
var args []reflect.Value var args []reflect.Value
if len(request.Params) > 0 { if len(request.Params) > 0 {
args, err = jsonParamsToArgsRPC(rpcFunc, request.Params)
args, err = jsonParamsToArgsRPC(rpcFunc, cdc, request.Params)
if err != nil { if err != nil {
WriteRPCResponseHTTP(w, types.RPCInvalidParamsError(request.ID, errors.Wrap(err, "Error converting json params to arguments"))) WriteRPCResponseHTTP(w, types.RPCInvalidParamsError(request.ID, errors.Wrap(err, "Error converting json params to arguments")))
return return
@ -148,18 +149,18 @@ func makeJSONRPCHandler(funcMap map[string]*RPCFunc, logger log.Logger) http.Han
WriteRPCResponseHTTP(w, types.RPCInternalError(request.ID, err)) WriteRPCResponseHTTP(w, types.RPCInternalError(request.ID, err))
return return
} }
WriteRPCResponseHTTP(w, types.NewRPCSuccessResponse(request.ID, result))
WriteRPCResponseHTTP(w, types.NewRPCSuccessResponse(cdc, request.ID, result))
} }
} }
func mapParamsToArgs(rpcFunc *RPCFunc, params map[string]*json.RawMessage, argsOffset int) ([]reflect.Value, error) {
func mapParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, params map[string]json.RawMessage, argsOffset int) ([]reflect.Value, error) {
values := make([]reflect.Value, len(rpcFunc.argNames)) values := make([]reflect.Value, len(rpcFunc.argNames))
for i, argName := range rpcFunc.argNames { for i, argName := range rpcFunc.argNames {
argType := rpcFunc.args[i+argsOffset] argType := rpcFunc.args[i+argsOffset]
if p, ok := params[argName]; ok && p != nil && len(*p) > 0 {
if p, ok := params[argName]; ok && p != nil && len(p) > 0 {
val := reflect.New(argType) val := reflect.New(argType)
err := json.Unmarshal(*p, val.Interface())
err := cdc.UnmarshalJSON(p, val.Interface())
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -172,7 +173,7 @@ func mapParamsToArgs(rpcFunc *RPCFunc, params map[string]*json.RawMessage, argsO
return values, nil return values, nil
} }
func arrayParamsToArgs(rpcFunc *RPCFunc, params []*json.RawMessage, argsOffset int) ([]reflect.Value, error) {
func arrayParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, params []json.RawMessage, argsOffset int) ([]reflect.Value, error) {
if len(rpcFunc.argNames) != len(params) { if len(rpcFunc.argNames) != len(params) {
return nil, errors.Errorf("Expected %v parameters (%v), got %v (%v)", return nil, errors.Errorf("Expected %v parameters (%v), got %v (%v)",
len(rpcFunc.argNames), rpcFunc.argNames, len(params), params) len(rpcFunc.argNames), rpcFunc.argNames, len(params), params)
@ -182,7 +183,7 @@ func arrayParamsToArgs(rpcFunc *RPCFunc, params []*json.RawMessage, argsOffset i
for i, p := range params { for i, p := range params {
argType := rpcFunc.args[i+argsOffset] argType := rpcFunc.args[i+argsOffset]
val := reflect.New(argType) val := reflect.New(argType)
err := json.Unmarshal(*p, val.Interface())
err := cdc.UnmarshalJSON(p, val.Interface())
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -191,39 +192,41 @@ func arrayParamsToArgs(rpcFunc *RPCFunc, params []*json.RawMessage, argsOffset i
return values, nil return values, nil
} }
// raw is unparsed json (from json.RawMessage) encoding either a map or an array.
// `raw` is unparsed json (from json.RawMessage) encoding either a map or an array.
// `argsOffset` should be 0 for RPC calls, and 1 for WS requests, where len(rpcFunc.args) != len(rpcFunc.argNames).
// //
// argsOffset should be 0 for RPC calls, and 1 for WS requests, where len(rpcFunc.args) != len(rpcFunc.argNames).
// Example: // Example:
// rpcFunc.args = [rpctypes.WSRPCContext string] // rpcFunc.args = [rpctypes.WSRPCContext string]
// rpcFunc.argNames = ["arg"] // rpcFunc.argNames = ["arg"]
func jsonParamsToArgs(rpcFunc *RPCFunc, raw []byte, argsOffset int) ([]reflect.Value, error) {
// first, try to get the map..
var m map[string]*json.RawMessage
func jsonParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, raw []byte, argsOffset int) ([]reflect.Value, error) {
// TODO: Make more efficient, perhaps by checking the first character for '{' or '['?
// First, try to get the map.
var m map[string]json.RawMessage
err := json.Unmarshal(raw, &m) err := json.Unmarshal(raw, &m)
if err == nil { if err == nil {
return mapParamsToArgs(rpcFunc, m, argsOffset)
return mapParamsToArgs(rpcFunc, cdc, m, argsOffset)
} }
// otherwise, try an array
var a []*json.RawMessage
// Ttherwise, try an array.
var a []json.RawMessage
err = json.Unmarshal(raw, &a) err = json.Unmarshal(raw, &a)
if err == nil { if err == nil {
return arrayParamsToArgs(rpcFunc, a, argsOffset)
return arrayParamsToArgs(rpcFunc, cdc, a, argsOffset)
} }
// otherwise, bad format, we cannot parse
// Otherwise, bad format, we cannot parse
return nil, errors.Errorf("Unknown type for JSON params: %v. Expected map or array", err) return nil, errors.Errorf("Unknown type for JSON params: %v. Expected map or array", err)
} }
// Convert a []interface{} OR a map[string]interface{} to properly typed values // Convert a []interface{} OR a map[string]interface{} to properly typed values
func jsonParamsToArgsRPC(rpcFunc *RPCFunc, params json.RawMessage) ([]reflect.Value, error) {
return jsonParamsToArgs(rpcFunc, params, 0)
func jsonParamsToArgsRPC(rpcFunc *RPCFunc, cdc *amino.Codec, params json.RawMessage) ([]reflect.Value, error) {
return jsonParamsToArgs(rpcFunc, cdc, params, 0)
} }
// Same as above, but with the first param the websocket connection // Same as above, but with the first param the websocket connection
func jsonParamsToArgsWS(rpcFunc *RPCFunc, params json.RawMessage, wsCtx types.WSRPCContext) ([]reflect.Value, error) {
values, err := jsonParamsToArgs(rpcFunc, params, 1)
func jsonParamsToArgsWS(rpcFunc *RPCFunc, cdc *amino.Codec, params json.RawMessage, wsCtx types.WSRPCContext) ([]reflect.Value, error) {
values, err := jsonParamsToArgs(rpcFunc, cdc, params, 1)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -235,7 +238,7 @@ func jsonParamsToArgsWS(rpcFunc *RPCFunc, params json.RawMessage, wsCtx types.WS
// rpc.http // rpc.http
// convert from a function name to the http handler // convert from a function name to the http handler
func makeHTTPHandler(rpcFunc *RPCFunc, logger log.Logger) func(http.ResponseWriter, *http.Request) {
func makeHTTPHandler(rpcFunc *RPCFunc, cdc *amino.Codec, logger log.Logger) func(http.ResponseWriter, *http.Request) {
// Exception for websocket endpoints // Exception for websocket endpoints
if rpcFunc.ws { if rpcFunc.ws {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
@ -245,7 +248,7 @@ func makeHTTPHandler(rpcFunc *RPCFunc, logger log.Logger) func(http.ResponseWrit
// All other endpoints // All other endpoints
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
logger.Debug("HTTP HANDLER", "req", r) logger.Debug("HTTP HANDLER", "req", r)
args, err := httpParamsToArgs(rpcFunc, r)
args, err := httpParamsToArgs(rpcFunc, cdc, r)
if err != nil { if err != nil {
WriteRPCResponseHTTP(w, types.RPCInvalidParamsError("", errors.Wrap(err, "Error converting http params to arguments"))) WriteRPCResponseHTTP(w, types.RPCInvalidParamsError("", errors.Wrap(err, "Error converting http params to arguments")))
return return
@ -257,13 +260,13 @@ func makeHTTPHandler(rpcFunc *RPCFunc, logger log.Logger) func(http.ResponseWrit
WriteRPCResponseHTTP(w, types.RPCInternalError("", err)) WriteRPCResponseHTTP(w, types.RPCInternalError("", err))
return return
} }
WriteRPCResponseHTTP(w, types.NewRPCSuccessResponse("", result))
WriteRPCResponseHTTP(w, types.NewRPCSuccessResponse(cdc, "", result))
} }
} }
// Covert an http query to a list of properly typed values. // 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). // 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) {
func httpParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, r *http.Request) ([]reflect.Value, error) {
values := make([]reflect.Value, len(rpcFunc.args)) values := make([]reflect.Value, len(rpcFunc.args))
for i, name := range rpcFunc.argNames { for i, name := range rpcFunc.argNames {
@ -278,7 +281,7 @@ func httpParamsToArgs(rpcFunc *RPCFunc, r *http.Request) ([]reflect.Value, error
continue continue
} }
v, err, ok := nonJsonToArg(argType, arg)
v, err, ok := nonJSONToArg(cdc, argType, arg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -287,7 +290,7 @@ func httpParamsToArgs(rpcFunc *RPCFunc, r *http.Request) ([]reflect.Value, error
continue continue
} }
values[i], err = _jsonStringToArg(argType, arg)
values[i], err = _jsonStringToArg(cdc, argType, arg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -296,9 +299,9 @@ func httpParamsToArgs(rpcFunc *RPCFunc, r *http.Request) ([]reflect.Value, error
return values, nil return values, nil
} }
func _jsonStringToArg(ty reflect.Type, arg string) (reflect.Value, error) {
func _jsonStringToArg(cdc *amino.Codec, ty reflect.Type, arg string) (reflect.Value, error) {
v := reflect.New(ty) v := reflect.New(ty)
err := json.Unmarshal([]byte(arg), v.Interface())
err := cdc.UnmarshalJSON([]byte(arg), v.Interface())
if err != nil { if err != nil {
return v, err return v, err
} }
@ -306,7 +309,7 @@ func _jsonStringToArg(ty reflect.Type, arg string) (reflect.Value, error) {
return v, nil return v, nil
} }
func nonJsonToArg(ty reflect.Type, arg string) (reflect.Value, error, bool) {
func nonJSONToArg(cdc *amino.Codec, ty reflect.Type, arg string) (reflect.Value, error, bool) {
isQuotedString := strings.HasPrefix(arg, `"`) && strings.HasSuffix(arg, `"`) isQuotedString := strings.HasPrefix(arg, `"`) && strings.HasSuffix(arg, `"`)
isHexString := strings.HasPrefix(strings.ToLower(arg), "0x") isHexString := strings.HasPrefix(strings.ToLower(arg), "0x")
expectingString := ty.Kind() == reflect.String expectingString := ty.Kind() == reflect.String
@ -332,7 +335,7 @@ func nonJsonToArg(ty reflect.Type, arg string) (reflect.Value, error, bool) {
if isQuotedString && expectingByteSlice { if isQuotedString && expectingByteSlice {
v := reflect.New(reflect.TypeOf("")) v := reflect.New(reflect.TypeOf(""))
err := json.Unmarshal([]byte(arg), v.Interface())
err := cdc.UnmarshalJSON([]byte(arg), v.Interface())
if err != nil { if err != nil {
return reflect.ValueOf(nil), err, false return reflect.ValueOf(nil), err, false
} }
@ -366,6 +369,7 @@ type wsConnection struct {
writeChan chan types.RPCResponse writeChan chan types.RPCResponse
funcMap map[string]*RPCFunc funcMap map[string]*RPCFunc
cdc *amino.Codec
// write channel capacity // write channel capacity
writeChanCapacity int writeChanCapacity int
@ -389,11 +393,12 @@ type wsConnection struct {
// description of how to configure ping period and pong wait time. NOTE: if the // description of how to configure ping period and pong wait time. NOTE: if the
// write buffer is full, pongs may be dropped, which may cause clients to // write buffer is full, pongs may be dropped, which may cause clients to
// disconnect. see https://github.com/gorilla/websocket/issues/97 // disconnect. see https://github.com/gorilla/websocket/issues/97
func NewWSConnection(baseConn *websocket.Conn, funcMap map[string]*RPCFunc, options ...func(*wsConnection)) *wsConnection {
func NewWSConnection(baseConn *websocket.Conn, funcMap map[string]*RPCFunc, cdc *amino.Codec, options ...func(*wsConnection)) *wsConnection {
wsc := &wsConnection{ wsc := &wsConnection{
remoteAddr: baseConn.RemoteAddr().String(), remoteAddr: baseConn.RemoteAddr().String(),
baseConn: baseConn, baseConn: baseConn,
funcMap: funcMap, funcMap: funcMap,
cdc: cdc,
writeWait: defaultWSWriteWait, writeWait: defaultWSWriteWait,
writeChanCapacity: defaultWSWriteChanCapacity, writeChanCapacity: defaultWSWriteChanCapacity,
readWait: defaultWSReadWait, readWait: defaultWSReadWait,
@ -569,11 +574,11 @@ func (wsc *wsConnection) readRoutine() {
if rpcFunc.ws { if rpcFunc.ws {
wsCtx := types.WSRPCContext{Request: request, WSRPCConnection: wsc} wsCtx := types.WSRPCContext{Request: request, WSRPCConnection: wsc}
if len(request.Params) > 0 { if len(request.Params) > 0 {
args, err = jsonParamsToArgsWS(rpcFunc, request.Params, wsCtx)
args, err = jsonParamsToArgsWS(rpcFunc, wsc.cdc, request.Params, wsCtx)
} }
} else { } else {
if len(request.Params) > 0 { if len(request.Params) > 0 {
args, err = jsonParamsToArgsRPC(rpcFunc, request.Params)
args, err = jsonParamsToArgsRPC(rpcFunc, wsc.cdc, request.Params)
} }
} }
if err != nil { if err != nil {
@ -590,7 +595,7 @@ func (wsc *wsConnection) readRoutine() {
wsc.WriteRPCResponse(types.RPCInternalError(request.ID, err)) wsc.WriteRPCResponse(types.RPCInternalError(request.ID, err))
continue continue
} else { } else {
wsc.WriteRPCResponse(types.NewRPCSuccessResponse(request.ID, result))
wsc.WriteRPCResponse(types.NewRPCSuccessResponse(wsc.cdc, request.ID, result))
continue continue
} }
@ -666,6 +671,7 @@ func (wsc *wsConnection) writeMessageWithDeadline(msgType int, msg []byte) error
type WebsocketManager struct { type WebsocketManager struct {
websocket.Upgrader websocket.Upgrader
funcMap map[string]*RPCFunc funcMap map[string]*RPCFunc
cdc *amino.Codec
logger log.Logger logger log.Logger
wsConnOptions []func(*wsConnection) wsConnOptions []func(*wsConnection)
} }
@ -673,9 +679,10 @@ type WebsocketManager struct {
// NewWebsocketManager returns a new WebsocketManager that routes according to // NewWebsocketManager returns a new WebsocketManager that routes according to
// the given funcMap and connects to the server with the given connection // the given funcMap and connects to the server with the given connection
// options. // options.
func NewWebsocketManager(funcMap map[string]*RPCFunc, wsConnOptions ...func(*wsConnection)) *WebsocketManager {
func NewWebsocketManager(funcMap map[string]*RPCFunc, cdc *amino.Codec, wsConnOptions ...func(*wsConnection)) *WebsocketManager {
return &WebsocketManager{ return &WebsocketManager{
funcMap: funcMap, funcMap: funcMap,
cdc: cdc,
Upgrader: websocket.Upgrader{ Upgrader: websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { CheckOrigin: func(r *http.Request) bool {
// TODO ??? // TODO ???
@ -702,7 +709,7 @@ func (wm *WebsocketManager) WebsocketHandler(w http.ResponseWriter, r *http.Requ
} }
// register connection // register connection
con := NewWSConnection(wsConn, wm.funcMap, wm.wsConnOptions...)
con := NewWSConnection(wsConn, wm.funcMap, wm.cdc, wm.wsConnOptions...)
con.SetLogger(wm.logger.With("remote", wsConn.RemoteAddr())) con.SetLogger(wm.logger.With("remote", wsConn.RemoteAddr()))
wm.logger.Info("New websocket connection", "remote", con.remoteAddr) wm.logger.Info("New websocket connection", "remote", con.remoteAddr)
err = con.Start() // Blocking err = con.Start() // Blocking


+ 3
- 1
rpc/lib/server/handlers_test.go View File

@ -12,6 +12,7 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/tendermint/go-amino"
rs "github.com/tendermint/tendermint/rpc/lib/server" rs "github.com/tendermint/tendermint/rpc/lib/server"
types "github.com/tendermint/tendermint/rpc/lib/types" types "github.com/tendermint/tendermint/rpc/lib/types"
"github.com/tendermint/tmlibs/log" "github.com/tendermint/tmlibs/log"
@ -21,10 +22,11 @@ func testMux() *http.ServeMux {
funcMap := map[string]*rs.RPCFunc{ funcMap := map[string]*rs.RPCFunc{
"c": rs.NewRPCFunc(func(s string, i int) (string, error) { return "foo", nil }, "s,i"), "c": rs.NewRPCFunc(func(s string, i int) (string, error) { return "foo", nil }, "s,i"),
} }
cdc := amino.NewCodec()
mux := http.NewServeMux() mux := http.NewServeMux()
buf := new(bytes.Buffer) buf := new(bytes.Buffer)
logger := log.NewTMLogger(buf) logger := log.NewTMLogger(buf)
rs.RegisterRPCFuncs(mux, funcMap, logger)
rs.RegisterRPCFuncs(mux, funcMap, cdc, logger)
return mux return mux
} }


+ 3
- 1
rpc/lib/server/parse_test.go View File

@ -6,6 +6,7 @@ import (
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/tendermint/go-amino"
cmn "github.com/tendermint/tmlibs/common" cmn "github.com/tendermint/tmlibs/common"
) )
@ -138,6 +139,7 @@ func TestParseRPC(t *testing.T) {
demo := func(height int, name string) {} demo := func(height int, name string) {}
call := NewRPCFunc(demo, "height,name") call := NewRPCFunc(demo, "height,name")
cdc := amino.NewCodec()
cases := []struct { cases := []struct {
raw string raw string
@ -158,7 +160,7 @@ func TestParseRPC(t *testing.T) {
for idx, tc := range cases { for idx, tc := range cases {
i := strconv.Itoa(idx) i := strconv.Itoa(idx)
data := []byte(tc.raw) data := []byte(tc.raw)
vals, err := jsonParamsToArgs(call, data, 0)
vals, err := jsonParamsToArgs(call, cdc, data, 0)
if tc.fail { if tc.fail {
assert.NotNil(err, i) assert.NotNil(err, i)
} else { } else {


+ 3
- 1
rpc/lib/test/main.go View File

@ -5,6 +5,7 @@ import (
"net/http" "net/http"
"os" "os"
"github.com/tendermint/go-amino"
rpcserver "github.com/tendermint/tendermint/rpc/lib/server" rpcserver "github.com/tendermint/tendermint/rpc/lib/server"
cmn "github.com/tendermint/tmlibs/common" cmn "github.com/tendermint/tmlibs/common"
"github.com/tendermint/tmlibs/log" "github.com/tendermint/tmlibs/log"
@ -24,8 +25,9 @@ type Result struct {
func main() { func main() {
mux := http.NewServeMux() mux := http.NewServeMux()
cdc := amino.NewCodec()
logger := log.NewTMLogger(log.NewSyncWriter(os.Stdout)) logger := log.NewTMLogger(log.NewSyncWriter(os.Stdout))
rpcserver.RegisterRPCFuncs(mux, routes, logger)
rpcserver.RegisterRPCFuncs(mux, routes, cdc, logger)
_, err := rpcserver.StartHTTPServer("0.0.0.0:8008", mux, logger) _, err := rpcserver.StartHTTPServer("0.0.0.0:8008", mux, logger)
if err != nil { if err != nil {
cmn.Exit(err.Error()) cmn.Exit(err.Error())


+ 23
- 6
rpc/lib/types/types.go View File

@ -7,6 +7,7 @@ import (
"strings" "strings"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/tendermint/go-amino"
tmpubsub "github.com/tendermint/tmlibs/pubsub" tmpubsub "github.com/tendermint/tmlibs/pubsub"
) )
@ -33,8 +34,16 @@ func (req RPCRequest) String() string {
return fmt.Sprintf("[%s %s]", req.ID, req.Method) return fmt.Sprintf("[%s %s]", req.ID, req.Method)
} }
func MapToRequest(id string, method string, params map[string]interface{}) (RPCRequest, error) {
payload, err := json.Marshal(params)
func MapToRequest(cdc *amino.Codec, id string, method string, params map[string]interface{}) (RPCRequest, error) {
var params_ = make(map[string]json.RawMessage, len(params))
for name, value := range params {
valueJSON, err := cdc.MarshalJSON(value)
if err != nil {
return RPCRequest{}, err
}
params_[name] = valueJSON
}
payload, err := json.Marshal(params_) // NOTE: Amino doesn't handle maps yet.
if err != nil { if err != nil {
return RPCRequest{}, err return RPCRequest{}, err
} }
@ -42,8 +51,16 @@ func MapToRequest(id string, method string, params map[string]interface{}) (RPCR
return request, nil return request, nil
} }
func ArrayToRequest(id string, method string, params []interface{}) (RPCRequest, error) {
payload, err := json.Marshal(params)
func ArrayToRequest(cdc *amino.Codec, id string, method string, params []interface{}) (RPCRequest, error) {
var params_ = make([]json.RawMessage, len(params))
for i, value := range params {
valueJSON, err := cdc.MarshalJSON(value)
if err != nil {
return RPCRequest{}, err
}
params_[i] = valueJSON
}
payload, err := json.Marshal(params_) // NOTE: Amino doesn't handle maps yet.
if err != nil { if err != nil {
return RPCRequest{}, err return RPCRequest{}, err
} }
@ -75,12 +92,12 @@ type RPCResponse struct {
Error *RPCError `json:"error,omitempty"` Error *RPCError `json:"error,omitempty"`
} }
func NewRPCSuccessResponse(id string, res interface{}) RPCResponse {
func NewRPCSuccessResponse(cdc *amino.Codec, id string, res interface{}) RPCResponse {
var rawMsg json.RawMessage var rawMsg json.RawMessage
if res != nil { if res != nil {
var js []byte var js []byte
js, err := json.Marshal(res)
js, err := cdc.MarshalJSON(res)
if err != nil { if err != nil {
return RPCInternalError(id, errors.Wrap(err, "Error marshalling response")) return RPCInternalError(id, errors.Wrap(err, "Error marshalling response"))
} }


+ 3
- 1
rpc/lib/types/types_test.go View File

@ -8,6 +8,7 @@ import (
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/tendermint/go-amino"
) )
type SampleResult struct { type SampleResult struct {
@ -16,8 +17,9 @@ type SampleResult struct {
func TestResponses(t *testing.T) { func TestResponses(t *testing.T) {
assert := assert.New(t) assert := assert.New(t)
cdc := amino.NewCodec()
a := NewRPCSuccessResponse("1", &SampleResult{"hello"})
a := NewRPCSuccessResponse(cdc, "1", &SampleResult{"hello"})
b, _ := json.Marshal(a) b, _ := json.Marshal(a)
s := `{"jsonrpc":"2.0","id":"1","result":{"Value":"hello"}}` s := `{"jsonrpc":"2.0","id":"1","result":{"Value":"hello"}}`
assert.Equal(string(s), string(b)) assert.Equal(string(s), string(b))


Loading…
Cancel
Save