You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

782 lines
24 KiB

7 years ago
8 years ago
8 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
8 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
8 years ago
7 years ago
8 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
new pubsub package comment out failing consensus tests for now rewrite rpc httpclient to use new pubsub package import pubsub as tmpubsub, query as tmquery make event IDs constants EventKey -> EventTypeKey rename EventsPubsub to PubSub mempool does not use pubsub rename eventsSub to pubsub new subscribe API fix channel size issues and consensus tests bugs refactor rpc client add missing discardFromChan method add mutex rename pubsub to eventBus remove IsRunning from WSRPCConnection interface (not needed) add a comment in broadcastNewRoundStepsAndVotes rename registerEventCallbacks to broadcastNewRoundStepsAndVotes See https://dave.cheney.net/2014/03/19/channel-axioms stop eventBuses after reactor tests remove unnecessary Unsubscribe return subscribe helper function move discardFromChan to where it is used subscribe now returns an err this gives us ability to refuse to subscribe if pubsub is at its max capacity. use context for control overflow cache queries handle err when subscribing in replay_test rename testClientID to testSubscriber extract var set channel buffer capacity to 1 in replay_file fix byzantine_test unsubscribe from single event, not all events refactor httpclient to return events to appropriate channels return failing testReplayCrashBeforeWriteVote test fix TestValidatorSetChanges refactor code a bit fix testReplayCrashBeforeWriteVote add comment fix TestValidatorSetChanges fixes from Bucky's review update comment [ci skip] test TxEventBuffer update changelog fix TestValidatorSetChanges (2nd attempt) only do wg.Done when no errors benchmark event bus create pubsub server inside NewEventBus only expose config params (later if needed) set buffer capacity to 0 so we are not testing cache new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ} This should allow to subscribe to all transactions! or a specific one using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'" use TimeoutCommit instead of afterPublishEventNewBlockTimeout TimeoutCommit is the time a node waits after committing a block, before it goes into the next height. So it will finish everything from the last block, but then wait a bit. The idea is this gives it time to hear more votes from other validators, to strengthen the commit it includes in the next block. But it also gives it time to hear about new transactions. waitForBlockWithUpdatedVals rewrite WAL crash tests Task: test that we can recover from any WAL crash. Solution: the old tests were relying on event hub being run in the same thread (we were injecting the private validator's last signature). when considering a rewrite, we considered two possible solutions: write a "fuzzy" testing system where WAL is crashing upon receiving a new message, or inject failures and trigger them in tests using something like https://github.com/coreos/gofail. remove sleep no cs.Lock around wal.Save test different cases (empty block, non-empty block, ...) comments add comments test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks fixes as per Bucky's last review reset subscriptions on UnsubscribeAll use a simple counter to track message for which we panicked also, set a smaller part size for all test cases
7 years ago
7 years ago
7 years ago
8 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
new pubsub package comment out failing consensus tests for now rewrite rpc httpclient to use new pubsub package import pubsub as tmpubsub, query as tmquery make event IDs constants EventKey -> EventTypeKey rename EventsPubsub to PubSub mempool does not use pubsub rename eventsSub to pubsub new subscribe API fix channel size issues and consensus tests bugs refactor rpc client add missing discardFromChan method add mutex rename pubsub to eventBus remove IsRunning from WSRPCConnection interface (not needed) add a comment in broadcastNewRoundStepsAndVotes rename registerEventCallbacks to broadcastNewRoundStepsAndVotes See https://dave.cheney.net/2014/03/19/channel-axioms stop eventBuses after reactor tests remove unnecessary Unsubscribe return subscribe helper function move discardFromChan to where it is used subscribe now returns an err this gives us ability to refuse to subscribe if pubsub is at its max capacity. use context for control overflow cache queries handle err when subscribing in replay_test rename testClientID to testSubscriber extract var set channel buffer capacity to 1 in replay_file fix byzantine_test unsubscribe from single event, not all events refactor httpclient to return events to appropriate channels return failing testReplayCrashBeforeWriteVote test fix TestValidatorSetChanges refactor code a bit fix testReplayCrashBeforeWriteVote add comment fix TestValidatorSetChanges fixes from Bucky's review update comment [ci skip] test TxEventBuffer update changelog fix TestValidatorSetChanges (2nd attempt) only do wg.Done when no errors benchmark event bus create pubsub server inside NewEventBus only expose config params (later if needed) set buffer capacity to 0 so we are not testing cache new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ} This should allow to subscribe to all transactions! or a specific one using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'" use TimeoutCommit instead of afterPublishEventNewBlockTimeout TimeoutCommit is the time a node waits after committing a block, before it goes into the next height. So it will finish everything from the last block, but then wait a bit. The idea is this gives it time to hear more votes from other validators, to strengthen the commit it includes in the next block. But it also gives it time to hear about new transactions. waitForBlockWithUpdatedVals rewrite WAL crash tests Task: test that we can recover from any WAL crash. Solution: the old tests were relying on event hub being run in the same thread (we were injecting the private validator's last signature). when considering a rewrite, we considered two possible solutions: write a "fuzzy" testing system where WAL is crashing upon receiving a new message, or inject failures and trigger them in tests using something like https://github.com/coreos/gofail. remove sleep no cs.Lock around wal.Save test different cases (empty block, non-empty block, ...) comments add comments test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks fixes as per Bucky's last review reset subscriptions on UnsubscribeAll use a simple counter to track message for which we panicked also, set a smaller part size for all test cases
7 years ago
7 years ago
7 years ago
7 years ago
8 years ago
7 years ago
7 years ago
8 years ago
  1. package rpcserver
  2. import (
  3. "bytes"
  4. "context"
  5. "encoding/hex"
  6. "encoding/json"
  7. "fmt"
  8. "io/ioutil"
  9. "net/http"
  10. "reflect"
  11. "runtime/debug"
  12. "sort"
  13. "strings"
  14. "time"
  15. "github.com/gorilla/websocket"
  16. "github.com/pkg/errors"
  17. "github.com/tendermint/go-amino"
  18. types "github.com/tendermint/tendermint/rpc/lib/types"
  19. cmn "github.com/tendermint/tmlibs/common"
  20. "github.com/tendermint/tmlibs/log"
  21. )
  22. // RegisterRPCFuncs adds a route for each function in the funcMap, as well as general jsonrpc and websocket handlers for all functions.
  23. // "result" is the interface on which the result objects are registered, and is popualted with every RPCResponse
  24. func RegisterRPCFuncs(mux *http.ServeMux, funcMap map[string]*RPCFunc, cdc *amino.Codec, logger log.Logger) {
  25. // HTTP endpoints
  26. for funcName, rpcFunc := range funcMap {
  27. mux.HandleFunc("/"+funcName, makeHTTPHandler(rpcFunc, cdc, logger))
  28. }
  29. // JSONRPC endpoints
  30. mux.HandleFunc("/", makeJSONRPCHandler(funcMap, cdc, logger))
  31. }
  32. //-------------------------------------
  33. // function introspection
  34. // RPCFunc contains the introspected type information for a function
  35. type RPCFunc struct {
  36. f reflect.Value // underlying rpc function
  37. args []reflect.Type // type of each function arg
  38. returns []reflect.Type // type of each return arg
  39. argNames []string // name of each argument
  40. ws bool // websocket only
  41. }
  42. // NewRPCFunc wraps a function for introspection.
  43. // f is the function, args are comma separated argument names
  44. func NewRPCFunc(f interface{}, args string) *RPCFunc {
  45. return newRPCFunc(f, args, false)
  46. }
  47. // NewWSRPCFunc wraps a function for introspection and use in the websockets.
  48. func NewWSRPCFunc(f interface{}, args string) *RPCFunc {
  49. return newRPCFunc(f, args, true)
  50. }
  51. func newRPCFunc(f interface{}, args string, ws bool) *RPCFunc {
  52. var argNames []string
  53. if args != "" {
  54. argNames = strings.Split(args, ",")
  55. }
  56. return &RPCFunc{
  57. f: reflect.ValueOf(f),
  58. args: funcArgTypes(f),
  59. returns: funcReturnTypes(f),
  60. argNames: argNames,
  61. ws: ws,
  62. }
  63. }
  64. // return a function's argument types
  65. func funcArgTypes(f interface{}) []reflect.Type {
  66. t := reflect.TypeOf(f)
  67. n := t.NumIn()
  68. typez := make([]reflect.Type, n)
  69. for i := 0; i < n; i++ {
  70. typez[i] = t.In(i)
  71. }
  72. return typez
  73. }
  74. // return a function's return types
  75. func funcReturnTypes(f interface{}) []reflect.Type {
  76. t := reflect.TypeOf(f)
  77. n := t.NumOut()
  78. typez := make([]reflect.Type, n)
  79. for i := 0; i < n; i++ {
  80. typez[i] = t.Out(i)
  81. }
  82. return typez
  83. }
  84. // function introspection
  85. //-----------------------------------------------------------------------------
  86. // rpc.json
  87. // jsonrpc calls grab the given method's function info and runs reflect.Call
  88. func makeJSONRPCHandler(funcMap map[string]*RPCFunc, cdc *amino.Codec, logger log.Logger) http.HandlerFunc {
  89. return func(w http.ResponseWriter, r *http.Request) {
  90. b, err := ioutil.ReadAll(r.Body)
  91. if err != nil {
  92. WriteRPCResponseHTTP(w, types.RPCInvalidRequestError("", errors.Wrap(err, "Error reading request body")))
  93. return
  94. }
  95. // if its an empty request (like from a browser),
  96. // just display a list of functions
  97. if len(b) == 0 {
  98. writeListOfEndpoints(w, r, funcMap)
  99. return
  100. }
  101. var request types.RPCRequest
  102. err = json.Unmarshal(b, &request)
  103. if err != nil {
  104. WriteRPCResponseHTTP(w, types.RPCParseError("", errors.Wrap(err, "Error unmarshalling request")))
  105. return
  106. }
  107. // A Notification is a Request object without an "id" member.
  108. // The Server MUST NOT reply to a Notification, including those that are within a batch request.
  109. if request.ID == "" {
  110. logger.Debug("HTTPJSONRPC received a notification, skipping... (please send a non-empty ID if you want to call a method)")
  111. return
  112. }
  113. if len(r.URL.Path) > 1 {
  114. WriteRPCResponseHTTP(w, types.RPCInvalidRequestError(request.ID, errors.Errorf("Path %s is invalid", r.URL.Path)))
  115. return
  116. }
  117. rpcFunc := funcMap[request.Method]
  118. if rpcFunc == nil || rpcFunc.ws {
  119. WriteRPCResponseHTTP(w, types.RPCMethodNotFoundError(request.ID))
  120. return
  121. }
  122. var args []reflect.Value
  123. if len(request.Params) > 0 {
  124. args, err = jsonParamsToArgsRPC(rpcFunc, cdc, request.Params)
  125. if err != nil {
  126. WriteRPCResponseHTTP(w, types.RPCInvalidParamsError(request.ID, errors.Wrap(err, "Error converting json params to arguments")))
  127. return
  128. }
  129. }
  130. returns := rpcFunc.f.Call(args)
  131. logger.Info("HTTPJSONRPC", "method", request.Method, "args", args, "returns", returns)
  132. result, err := unreflectResult(returns)
  133. if err != nil {
  134. WriteRPCResponseHTTP(w, types.RPCInternalError(request.ID, err))
  135. return
  136. }
  137. WriteRPCResponseHTTP(w, types.NewRPCSuccessResponse(cdc, request.ID, result))
  138. }
  139. }
  140. func mapParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, params map[string]json.RawMessage, argsOffset int) ([]reflect.Value, error) {
  141. values := make([]reflect.Value, len(rpcFunc.argNames))
  142. for i, argName := range rpcFunc.argNames {
  143. argType := rpcFunc.args[i+argsOffset]
  144. if p, ok := params[argName]; ok && p != nil && len(p) > 0 {
  145. val := reflect.New(argType)
  146. err := cdc.UnmarshalJSON(p, val.Interface())
  147. if err != nil {
  148. return nil, err
  149. }
  150. values[i] = val.Elem()
  151. } else { // use default for that type
  152. values[i] = reflect.Zero(argType)
  153. }
  154. }
  155. return values, nil
  156. }
  157. func arrayParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, params []json.RawMessage, argsOffset int) ([]reflect.Value, error) {
  158. if len(rpcFunc.argNames) != len(params) {
  159. return nil, errors.Errorf("Expected %v parameters (%v), got %v (%v)",
  160. len(rpcFunc.argNames), rpcFunc.argNames, len(params), params)
  161. }
  162. values := make([]reflect.Value, len(params))
  163. for i, p := range params {
  164. argType := rpcFunc.args[i+argsOffset]
  165. val := reflect.New(argType)
  166. err := cdc.UnmarshalJSON(p, val.Interface())
  167. if err != nil {
  168. return nil, err
  169. }
  170. values[i] = val.Elem()
  171. }
  172. return values, nil
  173. }
  174. // `raw` is unparsed json (from json.RawMessage) encoding either a map or an array.
  175. // `argsOffset` should be 0 for RPC calls, and 1 for WS requests, where len(rpcFunc.args) != len(rpcFunc.argNames).
  176. //
  177. // Example:
  178. // rpcFunc.args = [rpctypes.WSRPCContext string]
  179. // rpcFunc.argNames = ["arg"]
  180. func jsonParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, raw []byte, argsOffset int) ([]reflect.Value, error) {
  181. // TODO: Make more efficient, perhaps by checking the first character for '{' or '['?
  182. // First, try to get the map.
  183. var m map[string]json.RawMessage
  184. err := json.Unmarshal(raw, &m)
  185. if err == nil {
  186. return mapParamsToArgs(rpcFunc, cdc, m, argsOffset)
  187. }
  188. // Otherwise, try an array.
  189. var a []json.RawMessage
  190. err = json.Unmarshal(raw, &a)
  191. if err == nil {
  192. return arrayParamsToArgs(rpcFunc, cdc, a, argsOffset)
  193. }
  194. // Otherwise, bad format, we cannot parse
  195. return nil, errors.Errorf("Unknown type for JSON params: %v. Expected map or array", err)
  196. }
  197. // Convert a []interface{} OR a map[string]interface{} to properly typed values
  198. func jsonParamsToArgsRPC(rpcFunc *RPCFunc, cdc *amino.Codec, params json.RawMessage) ([]reflect.Value, error) {
  199. return jsonParamsToArgs(rpcFunc, cdc, params, 0)
  200. }
  201. // Same as above, but with the first param the websocket connection
  202. func jsonParamsToArgsWS(rpcFunc *RPCFunc, cdc *amino.Codec, params json.RawMessage, wsCtx types.WSRPCContext) ([]reflect.Value, error) {
  203. values, err := jsonParamsToArgs(rpcFunc, cdc, params, 1)
  204. if err != nil {
  205. return nil, err
  206. }
  207. return append([]reflect.Value{reflect.ValueOf(wsCtx)}, values...), nil
  208. }
  209. // rpc.json
  210. //-----------------------------------------------------------------------------
  211. // rpc.http
  212. // convert from a function name to the http handler
  213. func makeHTTPHandler(rpcFunc *RPCFunc, cdc *amino.Codec, logger log.Logger) func(http.ResponseWriter, *http.Request) {
  214. // Exception for websocket endpoints
  215. if rpcFunc.ws {
  216. return func(w http.ResponseWriter, r *http.Request) {
  217. WriteRPCResponseHTTP(w, types.RPCMethodNotFoundError(""))
  218. }
  219. }
  220. // All other endpoints
  221. return func(w http.ResponseWriter, r *http.Request) {
  222. logger.Debug("HTTP HANDLER", "req", r)
  223. args, err := httpParamsToArgs(rpcFunc, cdc, r)
  224. if err != nil {
  225. WriteRPCResponseHTTP(w, types.RPCInvalidParamsError("", errors.Wrap(err, "Error converting http params to arguments")))
  226. return
  227. }
  228. returns := rpcFunc.f.Call(args)
  229. logger.Info("HTTPRestRPC", "method", r.URL.Path, "args", args, "returns", returns)
  230. result, err := unreflectResult(returns)
  231. if err != nil {
  232. WriteRPCResponseHTTP(w, types.RPCInternalError("", err))
  233. return
  234. }
  235. WriteRPCResponseHTTP(w, types.NewRPCSuccessResponse(cdc, "", result))
  236. }
  237. }
  238. // Covert an http query to a list of properly typed values.
  239. // To be properly decoded the arg must be a concrete type from tendermint (if its an interface).
  240. func httpParamsToArgs(rpcFunc *RPCFunc, cdc *amino.Codec, r *http.Request) ([]reflect.Value, error) {
  241. values := make([]reflect.Value, len(rpcFunc.args))
  242. for i, name := range rpcFunc.argNames {
  243. argType := rpcFunc.args[i]
  244. values[i] = reflect.Zero(argType) // set default for that type
  245. arg := GetParam(r, name)
  246. // log.Notice("param to arg", "argType", argType, "name", name, "arg", arg)
  247. if "" == arg {
  248. continue
  249. }
  250. v, err, ok := nonJSONToArg(cdc, argType, arg)
  251. if err != nil {
  252. return nil, err
  253. }
  254. if ok {
  255. values[i] = v
  256. continue
  257. }
  258. values[i], err = _jsonStringToArg(cdc, argType, arg)
  259. if err != nil {
  260. return nil, err
  261. }
  262. }
  263. return values, nil
  264. }
  265. func _jsonStringToArg(cdc *amino.Codec, ty reflect.Type, arg string) (reflect.Value, error) {
  266. v := reflect.New(ty)
  267. err := cdc.UnmarshalJSON([]byte(arg), v.Interface())
  268. if err != nil {
  269. return v, err
  270. }
  271. v = v.Elem()
  272. return v, nil
  273. }
  274. func nonJSONToArg(cdc *amino.Codec, ty reflect.Type, arg string) (reflect.Value, error, bool) {
  275. isQuotedString := strings.HasPrefix(arg, `"`) && strings.HasSuffix(arg, `"`)
  276. isHexString := strings.HasPrefix(strings.ToLower(arg), "0x")
  277. expectingString := ty.Kind() == reflect.String
  278. expectingByteSlice := ty.Kind() == reflect.Slice && ty.Elem().Kind() == reflect.Uint8
  279. if isHexString {
  280. if !expectingString && !expectingByteSlice {
  281. err := errors.Errorf("Got a hex string arg, but expected '%s'",
  282. ty.Kind().String())
  283. return reflect.ValueOf(nil), err, false
  284. }
  285. var value []byte
  286. value, err := hex.DecodeString(arg[2:])
  287. if err != nil {
  288. return reflect.ValueOf(nil), err, false
  289. }
  290. if ty.Kind() == reflect.String {
  291. return reflect.ValueOf(string(value)), nil, true
  292. }
  293. return reflect.ValueOf([]byte(value)), nil, true
  294. }
  295. if isQuotedString && expectingByteSlice {
  296. v := reflect.New(reflect.TypeOf(""))
  297. err := cdc.UnmarshalJSON([]byte(arg), v.Interface())
  298. if err != nil {
  299. return reflect.ValueOf(nil), err, false
  300. }
  301. v = v.Elem()
  302. return reflect.ValueOf([]byte(v.String())), nil, true
  303. }
  304. return reflect.ValueOf(nil), nil, false
  305. }
  306. // rpc.http
  307. //-----------------------------------------------------------------------------
  308. // rpc.websocket
  309. const (
  310. defaultWSWriteChanCapacity = 1000
  311. defaultWSWriteWait = 10 * time.Second
  312. defaultWSReadWait = 30 * time.Second
  313. defaultWSPingPeriod = (defaultWSReadWait * 9) / 10
  314. )
  315. // A single websocket connection contains listener id, underlying ws
  316. // connection, and the event switch for subscribing to events.
  317. //
  318. // In case of an error, the connection is stopped.
  319. type wsConnection struct {
  320. cmn.BaseService
  321. remoteAddr string
  322. baseConn *websocket.Conn
  323. writeChan chan types.RPCResponse
  324. funcMap map[string]*RPCFunc
  325. cdc *amino.Codec
  326. // write channel capacity
  327. writeChanCapacity int
  328. // each write times out after this.
  329. writeWait time.Duration
  330. // Connection times out if we haven't received *anything* in this long, not even pings.
  331. readWait time.Duration
  332. // Send pings to server with this period. Must be less than readWait, but greater than zero.
  333. pingPeriod time.Duration
  334. // object that is used to subscribe / unsubscribe from events
  335. eventSub types.EventSubscriber
  336. }
  337. // NewWSConnection wraps websocket.Conn.
  338. //
  339. // See the commentary on the func(*wsConnection) functions for a detailed
  340. // description of how to configure ping period and pong wait time. NOTE: if the
  341. // write buffer is full, pongs may be dropped, which may cause clients to
  342. // disconnect. see https://github.com/gorilla/websocket/issues/97
  343. func NewWSConnection(baseConn *websocket.Conn, funcMap map[string]*RPCFunc, cdc *amino.Codec, options ...func(*wsConnection)) *wsConnection {
  344. wsc := &wsConnection{
  345. remoteAddr: baseConn.RemoteAddr().String(),
  346. baseConn: baseConn,
  347. funcMap: funcMap,
  348. cdc: cdc,
  349. writeWait: defaultWSWriteWait,
  350. writeChanCapacity: defaultWSWriteChanCapacity,
  351. readWait: defaultWSReadWait,
  352. pingPeriod: defaultWSPingPeriod,
  353. }
  354. for _, option := range options {
  355. option(wsc)
  356. }
  357. wsc.BaseService = *cmn.NewBaseService(nil, "wsConnection", wsc)
  358. return wsc
  359. }
  360. // EventSubscriber sets object that is used to subscribe / unsubscribe from
  361. // events - not Goroutine-safe. If none given, default node's eventBus will be
  362. // used.
  363. func EventSubscriber(eventSub types.EventSubscriber) func(*wsConnection) {
  364. return func(wsc *wsConnection) {
  365. wsc.eventSub = eventSub
  366. }
  367. }
  368. // WriteWait sets the amount of time to wait before a websocket write times out.
  369. // It should only be used in the constructor - not Goroutine-safe.
  370. func WriteWait(writeWait time.Duration) func(*wsConnection) {
  371. return func(wsc *wsConnection) {
  372. wsc.writeWait = writeWait
  373. }
  374. }
  375. // WriteChanCapacity sets the capacity of the websocket write channel.
  376. // It should only be used in the constructor - not Goroutine-safe.
  377. func WriteChanCapacity(cap int) func(*wsConnection) {
  378. return func(wsc *wsConnection) {
  379. wsc.writeChanCapacity = cap
  380. }
  381. }
  382. // ReadWait sets the amount of time to wait before a websocket read times out.
  383. // It should only be used in the constructor - not Goroutine-safe.
  384. func ReadWait(readWait time.Duration) func(*wsConnection) {
  385. return func(wsc *wsConnection) {
  386. wsc.readWait = readWait
  387. }
  388. }
  389. // PingPeriod sets the duration for sending websocket pings.
  390. // It should only be used in the constructor - not Goroutine-safe.
  391. func PingPeriod(pingPeriod time.Duration) func(*wsConnection) {
  392. return func(wsc *wsConnection) {
  393. wsc.pingPeriod = pingPeriod
  394. }
  395. }
  396. // OnStart implements cmn.Service by starting the read and write routines. It
  397. // blocks until the connection closes.
  398. func (wsc *wsConnection) OnStart() error {
  399. wsc.writeChan = make(chan types.RPCResponse, wsc.writeChanCapacity)
  400. // Read subscriptions/unsubscriptions to events
  401. go wsc.readRoutine()
  402. // Write responses, BLOCKING.
  403. wsc.writeRoutine()
  404. return nil
  405. }
  406. // OnStop implements cmn.Service by unsubscribing remoteAddr from all subscriptions.
  407. func (wsc *wsConnection) OnStop() {
  408. // Both read and write loops close the websocket connection when they exit their loops.
  409. // The writeChan is never closed, to allow WriteRPCResponse() to fail.
  410. if wsc.eventSub != nil {
  411. wsc.eventSub.UnsubscribeAll(context.TODO(), wsc.remoteAddr)
  412. }
  413. }
  414. // GetRemoteAddr returns the remote address of the underlying connection.
  415. // It implements WSRPCConnection
  416. func (wsc *wsConnection) GetRemoteAddr() string {
  417. return wsc.remoteAddr
  418. }
  419. // GetEventSubscriber implements WSRPCConnection by returning event subscriber.
  420. func (wsc *wsConnection) GetEventSubscriber() types.EventSubscriber {
  421. return wsc.eventSub
  422. }
  423. // WriteRPCResponse pushes a response to the writeChan, and blocks until it is accepted.
  424. // It implements WSRPCConnection. It is Goroutine-safe.
  425. func (wsc *wsConnection) WriteRPCResponse(resp types.RPCResponse) {
  426. select {
  427. case <-wsc.Quit():
  428. return
  429. case wsc.writeChan <- resp:
  430. }
  431. }
  432. // TryWriteRPCResponse attempts to push a response to the writeChan, but does not block.
  433. // It implements WSRPCConnection. It is Goroutine-safe
  434. func (wsc *wsConnection) TryWriteRPCResponse(resp types.RPCResponse) bool {
  435. select {
  436. case <-wsc.Quit():
  437. return false
  438. case wsc.writeChan <- resp:
  439. return true
  440. default:
  441. return false
  442. }
  443. }
  444. // Codec returns an amino codec used to decode parameters and encode results.
  445. // It implements WSRPCConnection.
  446. func (wsc *wsConnection) Codec() *amino.Codec {
  447. return wsc.cdc
  448. }
  449. // Read from the socket and subscribe to or unsubscribe from events
  450. func (wsc *wsConnection) readRoutine() {
  451. defer func() {
  452. if r := recover(); r != nil {
  453. err, ok := r.(error)
  454. if !ok {
  455. err = fmt.Errorf("WSJSONRPC: %v", r)
  456. }
  457. wsc.Logger.Error("Panic in WSJSONRPC handler", "err", err, "stack", string(debug.Stack()))
  458. wsc.WriteRPCResponse(types.RPCInternalError("unknown", err))
  459. go wsc.readRoutine()
  460. } else {
  461. wsc.baseConn.Close() // nolint: errcheck
  462. }
  463. }()
  464. wsc.baseConn.SetPongHandler(func(m string) error {
  465. return wsc.baseConn.SetReadDeadline(time.Now().Add(wsc.readWait))
  466. })
  467. for {
  468. select {
  469. case <-wsc.Quit():
  470. return
  471. default:
  472. // reset deadline for every type of message (control or data)
  473. if err := wsc.baseConn.SetReadDeadline(time.Now().Add(wsc.readWait)); err != nil {
  474. wsc.Logger.Error("failed to set read deadline", "err", err)
  475. }
  476. var in []byte
  477. _, in, err := wsc.baseConn.ReadMessage()
  478. if err != nil {
  479. if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
  480. wsc.Logger.Info("Client closed the connection")
  481. } else {
  482. wsc.Logger.Error("Failed to read request", "err", err)
  483. }
  484. wsc.Stop()
  485. return
  486. }
  487. var request types.RPCRequest
  488. err = json.Unmarshal(in, &request)
  489. if err != nil {
  490. wsc.WriteRPCResponse(types.RPCParseError("", errors.Wrap(err, "Error unmarshaling request")))
  491. continue
  492. }
  493. // A Notification is a Request object without an "id" member.
  494. // The Server MUST NOT reply to a Notification, including those that are within a batch request.
  495. if request.ID == "" {
  496. wsc.Logger.Debug("WSJSONRPC received a notification, skipping... (please send a non-empty ID if you want to call a method)")
  497. continue
  498. }
  499. // Now, fetch the RPCFunc and execute it.
  500. rpcFunc := wsc.funcMap[request.Method]
  501. if rpcFunc == nil {
  502. wsc.WriteRPCResponse(types.RPCMethodNotFoundError(request.ID))
  503. continue
  504. }
  505. var args []reflect.Value
  506. if rpcFunc.ws {
  507. wsCtx := types.WSRPCContext{Request: request, WSRPCConnection: wsc}
  508. if len(request.Params) > 0 {
  509. args, err = jsonParamsToArgsWS(rpcFunc, wsc.cdc, request.Params, wsCtx)
  510. }
  511. } else {
  512. if len(request.Params) > 0 {
  513. args, err = jsonParamsToArgsRPC(rpcFunc, wsc.cdc, request.Params)
  514. }
  515. }
  516. if err != nil {
  517. wsc.WriteRPCResponse(types.RPCInternalError(request.ID, errors.Wrap(err, "Error converting json params to arguments")))
  518. continue
  519. }
  520. returns := rpcFunc.f.Call(args)
  521. // TODO: Need to encode args/returns to string if we want to log them
  522. wsc.Logger.Info("WSJSONRPC", "method", request.Method)
  523. result, err := unreflectResult(returns)
  524. if err != nil {
  525. wsc.WriteRPCResponse(types.RPCInternalError(request.ID, err))
  526. continue
  527. } else {
  528. wsc.WriteRPCResponse(types.NewRPCSuccessResponse(wsc.cdc, request.ID, result))
  529. continue
  530. }
  531. }
  532. }
  533. }
  534. // receives on a write channel and writes out on the socket
  535. func (wsc *wsConnection) writeRoutine() {
  536. pingTicker := time.NewTicker(wsc.pingPeriod)
  537. defer func() {
  538. pingTicker.Stop()
  539. if err := wsc.baseConn.Close(); err != nil {
  540. wsc.Logger.Error("Error closing connection", "err", err)
  541. }
  542. }()
  543. // https://github.com/gorilla/websocket/issues/97
  544. pongs := make(chan string, 1)
  545. wsc.baseConn.SetPingHandler(func(m string) error {
  546. select {
  547. case pongs <- m:
  548. default:
  549. }
  550. return nil
  551. })
  552. for {
  553. select {
  554. case m := <-pongs:
  555. err := wsc.writeMessageWithDeadline(websocket.PongMessage, []byte(m))
  556. if err != nil {
  557. wsc.Logger.Info("Failed to write pong (client may disconnect)", "err", err)
  558. }
  559. case <-pingTicker.C:
  560. err := wsc.writeMessageWithDeadline(websocket.PingMessage, []byte{})
  561. if err != nil {
  562. wsc.Logger.Error("Failed to write ping", "err", err)
  563. wsc.Stop()
  564. return
  565. }
  566. case msg := <-wsc.writeChan:
  567. jsonBytes, err := json.MarshalIndent(msg, "", " ")
  568. if err != nil {
  569. wsc.Logger.Error("Failed to marshal RPCResponse to JSON", "err", err)
  570. } else {
  571. if err = wsc.writeMessageWithDeadline(websocket.TextMessage, jsonBytes); err != nil {
  572. wsc.Logger.Error("Failed to write response", "err", err)
  573. wsc.Stop()
  574. return
  575. }
  576. }
  577. case <-wsc.Quit():
  578. return
  579. }
  580. }
  581. }
  582. // All writes to the websocket must (re)set the write deadline.
  583. // If some writes don't set it while others do, they may timeout incorrectly (https://github.com/tendermint/tendermint/issues/553)
  584. func (wsc *wsConnection) writeMessageWithDeadline(msgType int, msg []byte) error {
  585. if err := wsc.baseConn.SetWriteDeadline(time.Now().Add(wsc.writeWait)); err != nil {
  586. return err
  587. }
  588. return wsc.baseConn.WriteMessage(msgType, msg)
  589. }
  590. //----------------------------------------
  591. // WebsocketManager is the main manager for all websocket connections.
  592. // It holds the event switch and a map of functions for routing.
  593. // NOTE: The websocket path is defined externally, e.g. in node/node.go
  594. type WebsocketManager struct {
  595. websocket.Upgrader
  596. funcMap map[string]*RPCFunc
  597. cdc *amino.Codec
  598. logger log.Logger
  599. wsConnOptions []func(*wsConnection)
  600. }
  601. // NewWebsocketManager returns a new WebsocketManager that routes according to
  602. // the given funcMap and connects to the server with the given connection
  603. // options.
  604. func NewWebsocketManager(funcMap map[string]*RPCFunc, cdc *amino.Codec, wsConnOptions ...func(*wsConnection)) *WebsocketManager {
  605. return &WebsocketManager{
  606. funcMap: funcMap,
  607. cdc: cdc,
  608. Upgrader: websocket.Upgrader{
  609. CheckOrigin: func(r *http.Request) bool {
  610. // TODO ???
  611. return true
  612. },
  613. },
  614. logger: log.NewNopLogger(),
  615. wsConnOptions: wsConnOptions,
  616. }
  617. }
  618. // SetLogger sets the logger.
  619. func (wm *WebsocketManager) SetLogger(l log.Logger) {
  620. wm.logger = l
  621. }
  622. // WebsocketHandler upgrades the request/response (via http.Hijack) and starts the wsConnection.
  623. func (wm *WebsocketManager) WebsocketHandler(w http.ResponseWriter, r *http.Request) {
  624. wsConn, err := wm.Upgrade(w, r, nil)
  625. if err != nil {
  626. // TODO - return http error
  627. wm.logger.Error("Failed to upgrade to websocket connection", "err", err)
  628. return
  629. }
  630. // register connection
  631. con := NewWSConnection(wsConn, wm.funcMap, wm.cdc, wm.wsConnOptions...)
  632. con.SetLogger(wm.logger.With("remote", wsConn.RemoteAddr()))
  633. wm.logger.Info("New websocket connection", "remote", con.remoteAddr)
  634. err = con.Start() // Blocking
  635. if err != nil {
  636. wm.logger.Error("Error starting connection", "err", err)
  637. }
  638. }
  639. // rpc.websocket
  640. //-----------------------------------------------------------------------------
  641. // NOTE: assume returns is result struct and error. If error is not nil, return it
  642. func unreflectResult(returns []reflect.Value) (interface{}, error) {
  643. errV := returns[1]
  644. if errV.Interface() != nil {
  645. return nil, errors.Errorf("%v", errV.Interface())
  646. }
  647. rv := returns[0]
  648. // the result is a registered interface,
  649. // we need a pointer to it so we can marshal with type byte
  650. rvp := reflect.New(rv.Type())
  651. rvp.Elem().Set(rv)
  652. return rvp.Interface(), nil
  653. }
  654. // writes a list of available rpc endpoints as an html page
  655. func writeListOfEndpoints(w http.ResponseWriter, r *http.Request, funcMap map[string]*RPCFunc) {
  656. noArgNames := []string{}
  657. argNames := []string{}
  658. for name, funcData := range funcMap {
  659. if len(funcData.args) == 0 {
  660. noArgNames = append(noArgNames, name)
  661. } else {
  662. argNames = append(argNames, name)
  663. }
  664. }
  665. sort.Strings(noArgNames)
  666. sort.Strings(argNames)
  667. buf := new(bytes.Buffer)
  668. buf.WriteString("<html><body>")
  669. buf.WriteString("<br>Available endpoints:<br>")
  670. for _, name := range noArgNames {
  671. link := fmt.Sprintf("//%s/%s", r.Host, name)
  672. buf.WriteString(fmt.Sprintf("<a href=\"%s\">%s</a></br>", link, link))
  673. }
  674. buf.WriteString("<br>Endpoints that require arguments:<br>")
  675. for _, name := range argNames {
  676. link := fmt.Sprintf("//%s/%s?", r.Host, name)
  677. funcData := funcMap[name]
  678. for i, argName := range funcData.argNames {
  679. link += argName + "=_"
  680. if i < len(funcData.argNames)-1 {
  681. link += "&"
  682. }
  683. }
  684. buf.WriteString(fmt.Sprintf("<a href=\"%s\">%s</a></br>", link, link))
  685. }
  686. buf.WriteString("</body></html>")
  687. w.Header().Set("Content-Type", "text/html")
  688. w.WriteHeader(200)
  689. w.Write(buf.Bytes()) // nolint: errcheck
  690. }