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.

342 lines
9.5 KiB

8 years ago
8 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
8 years ago
8 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
  1. package abcicli
  2. import (
  3. "net"
  4. "sync"
  5. "time"
  6. context "golang.org/x/net/context"
  7. grpc "google.golang.org/grpc"
  8. . "github.com/tendermint/go-common"
  9. "github.com/tendermint/abci/types"
  10. )
  11. // A stripped copy of the remoteClient that makes
  12. // synchronous calls using grpc
  13. type grpcClient struct {
  14. BaseService
  15. mustConnect bool
  16. client types.ABCIApplicationClient
  17. mtx sync.Mutex
  18. addr string
  19. err error
  20. resCb func(*types.Request, *types.Response) // listens to all callbacks
  21. }
  22. func NewGRPCClient(addr string, mustConnect bool) (*grpcClient, error) {
  23. cli := &grpcClient{
  24. addr: addr,
  25. mustConnect: mustConnect,
  26. }
  27. cli.BaseService = *NewBaseService(nil, "grpcClient", cli)
  28. _, err := cli.Start() // Just start it, it's confusing for callers to remember to start.
  29. return cli, err
  30. }
  31. func dialerFunc(addr string, timeout time.Duration) (net.Conn, error) {
  32. return Connect(addr)
  33. }
  34. func (cli *grpcClient) OnStart() error {
  35. cli.BaseService.OnStart()
  36. RETRY_LOOP:
  37. for {
  38. conn, err := grpc.Dial(cli.addr, grpc.WithInsecure(), grpc.WithDialer(dialerFunc))
  39. if err != nil {
  40. if cli.mustConnect {
  41. return err
  42. } else {
  43. log.Warn(Fmt("abci.grpcClient failed to connect to %v. Retrying...\n", cli.addr))
  44. time.Sleep(time.Second * 3)
  45. continue RETRY_LOOP
  46. }
  47. }
  48. client := types.NewABCIApplicationClient(conn)
  49. ENSURE_CONNECTED:
  50. for {
  51. _, err := client.Echo(context.Background(), &types.RequestEcho{"hello"}, grpc.FailFast(true))
  52. if err == nil {
  53. break ENSURE_CONNECTED
  54. }
  55. time.Sleep(time.Second)
  56. }
  57. cli.client = client
  58. return nil
  59. }
  60. }
  61. func (cli *grpcClient) OnStop() {
  62. cli.BaseService.OnStop()
  63. cli.mtx.Lock()
  64. defer cli.mtx.Unlock()
  65. // TODO: how to close conn? its not a net.Conn and grpc doesn't expose a Close()
  66. /*if cli.conn != nil {
  67. cli.conn.Close()
  68. }*/
  69. }
  70. func (cli *grpcClient) StopForError(err error) {
  71. cli.mtx.Lock()
  72. if !cli.IsRunning() {
  73. return
  74. }
  75. if cli.err == nil {
  76. cli.err = err
  77. }
  78. cli.mtx.Unlock()
  79. log.Warn(Fmt("Stopping abci.grpcClient for error: %v", err.Error()))
  80. cli.Stop()
  81. }
  82. func (cli *grpcClient) Error() error {
  83. cli.mtx.Lock()
  84. defer cli.mtx.Unlock()
  85. return cli.err
  86. }
  87. // Set listener for all responses
  88. // NOTE: callback may get internally generated flush responses.
  89. func (cli *grpcClient) SetResponseCallback(resCb Callback) {
  90. cli.mtx.Lock()
  91. defer cli.mtx.Unlock()
  92. cli.resCb = resCb
  93. }
  94. //----------------------------------------
  95. // GRPC calls are synchronous, but some callbacks expect to be called asynchronously
  96. // (eg. the mempool expects to be able to lock to remove bad txs from cache).
  97. // To accomodate, we finish each call in its own go-routine,
  98. // which is expensive, but easy - if you want something better, use the socket protocol!
  99. // maybe one day, if people really want it, we use grpc streams,
  100. // but hopefully not :D
  101. func (cli *grpcClient) EchoAsync(msg string) *ReqRes {
  102. req := types.ToRequestEcho(msg)
  103. res, err := cli.client.Echo(context.Background(), req.GetEcho(), grpc.FailFast(true))
  104. if err != nil {
  105. cli.StopForError(err)
  106. }
  107. return cli.finishAsyncCall(req, &types.Response{&types.Response_Echo{res}})
  108. }
  109. func (cli *grpcClient) FlushAsync() *ReqRes {
  110. req := types.ToRequestFlush()
  111. res, err := cli.client.Flush(context.Background(), req.GetFlush(), grpc.FailFast(true))
  112. if err != nil {
  113. cli.StopForError(err)
  114. }
  115. return cli.finishAsyncCall(req, &types.Response{&types.Response_Flush{res}})
  116. }
  117. func (cli *grpcClient) InfoAsync() *ReqRes {
  118. req := types.ToRequestInfo()
  119. res, err := cli.client.Info(context.Background(), req.GetInfo(), grpc.FailFast(true))
  120. if err != nil {
  121. cli.StopForError(err)
  122. }
  123. return cli.finishAsyncCall(req, &types.Response{&types.Response_Info{res}})
  124. }
  125. func (cli *grpcClient) SetOptionAsync(key string, value string) *ReqRes {
  126. req := types.ToRequestSetOption(key, value)
  127. res, err := cli.client.SetOption(context.Background(), req.GetSetOption(), grpc.FailFast(true))
  128. if err != nil {
  129. cli.StopForError(err)
  130. }
  131. return cli.finishAsyncCall(req, &types.Response{&types.Response_SetOption{res}})
  132. }
  133. func (cli *grpcClient) DeliverTxAsync(tx []byte) *ReqRes {
  134. req := types.ToRequestDeliverTx(tx)
  135. res, err := cli.client.DeliverTx(context.Background(), req.GetDeliverTx(), grpc.FailFast(true))
  136. if err != nil {
  137. cli.StopForError(err)
  138. }
  139. return cli.finishAsyncCall(req, &types.Response{&types.Response_DeliverTx{res}})
  140. }
  141. func (cli *grpcClient) CheckTxAsync(tx []byte) *ReqRes {
  142. req := types.ToRequestCheckTx(tx)
  143. res, err := cli.client.CheckTx(context.Background(), req.GetCheckTx(), grpc.FailFast(true))
  144. if err != nil {
  145. cli.StopForError(err)
  146. }
  147. return cli.finishAsyncCall(req, &types.Response{&types.Response_CheckTx{res}})
  148. }
  149. func (cli *grpcClient) QueryAsync(query []byte) *ReqRes {
  150. req := types.ToRequestQuery(query)
  151. res, err := cli.client.Query(context.Background(), req.GetQuery(), grpc.FailFast(true))
  152. if err != nil {
  153. cli.StopForError(err)
  154. }
  155. return cli.finishAsyncCall(req, &types.Response{&types.Response_Query{res}})
  156. }
  157. func (cli *grpcClient) CommitAsync() *ReqRes {
  158. req := types.ToRequestCommit()
  159. res, err := cli.client.Commit(context.Background(), req.GetCommit(), grpc.FailFast(true))
  160. if err != nil {
  161. cli.StopForError(err)
  162. }
  163. return cli.finishAsyncCall(req, &types.Response{&types.Response_Commit{res}})
  164. }
  165. func (cli *grpcClient) InitChainAsync(validators []*types.Validator) *ReqRes {
  166. req := types.ToRequestInitChain(validators)
  167. res, err := cli.client.InitChain(context.Background(), req.GetInitChain(), grpc.FailFast(true))
  168. if err != nil {
  169. cli.StopForError(err)
  170. }
  171. return cli.finishAsyncCall(req, &types.Response{&types.Response_InitChain{res}})
  172. }
  173. func (cli *grpcClient) BeginBlockAsync(hash []byte, header *types.Header) *ReqRes {
  174. req := types.ToRequestBeginBlock(hash, header)
  175. res, err := cli.client.BeginBlock(context.Background(), req.GetBeginBlock(), grpc.FailFast(true))
  176. if err != nil {
  177. cli.StopForError(err)
  178. }
  179. return cli.finishAsyncCall(req, &types.Response{&types.Response_BeginBlock{res}})
  180. }
  181. func (cli *grpcClient) EndBlockAsync(height uint64) *ReqRes {
  182. req := types.ToRequestEndBlock(height)
  183. res, err := cli.client.EndBlock(context.Background(), req.GetEndBlock(), grpc.FailFast(true))
  184. if err != nil {
  185. cli.StopForError(err)
  186. }
  187. return cli.finishAsyncCall(req, &types.Response{&types.Response_EndBlock{res}})
  188. }
  189. func (cli *grpcClient) finishAsyncCall(req *types.Request, res *types.Response) *ReqRes {
  190. reqres := NewReqRes(req)
  191. reqres.Response = res // Set response
  192. reqres.Done() // Release waiters
  193. reqres.SetDone() // so reqRes.SetCallback will run the callback
  194. // go routine for callbacks
  195. go func() {
  196. // Notify reqRes listener if set
  197. if cb := reqres.GetCallback(); cb != nil {
  198. cb(res)
  199. }
  200. // Notify client listener if set
  201. if cli.resCb != nil {
  202. cli.resCb(reqres.Request, res)
  203. }
  204. }()
  205. return reqres
  206. }
  207. func (cli *grpcClient) checkErrGetResult() types.Result {
  208. if err := cli.Error(); err != nil {
  209. // StopForError should already have been called if error is set
  210. return types.ErrInternalError.SetLog(err.Error())
  211. }
  212. return types.Result{}
  213. }
  214. //----------------------------------------
  215. func (cli *grpcClient) EchoSync(msg string) (res types.Result) {
  216. reqres := cli.EchoAsync(msg)
  217. if res := cli.checkErrGetResult(); res.IsErr() {
  218. return res
  219. }
  220. resp := reqres.Response.GetEcho()
  221. return types.NewResultOK([]byte(resp.Message), LOG)
  222. }
  223. func (cli *grpcClient) FlushSync() error {
  224. return nil
  225. }
  226. func (cli *grpcClient) InfoSync() (resInfo types.ResponseInfo, err error) {
  227. reqres := cli.InfoAsync()
  228. if err = cli.Error(); err != nil {
  229. return resInfo, err
  230. }
  231. if resInfo_ := reqres.Response.GetInfo(); resInfo_ != nil {
  232. return *resInfo_, nil
  233. } else {
  234. return resInfo, nil
  235. }
  236. }
  237. func (cli *grpcClient) SetOptionSync(key string, value string) (res types.Result) {
  238. reqres := cli.SetOptionAsync(key, value)
  239. if res := cli.checkErrGetResult(); res.IsErr() {
  240. return res
  241. }
  242. resp := reqres.Response.GetSetOption()
  243. return types.Result{Code: OK, Data: nil, Log: resp.Log}
  244. }
  245. func (cli *grpcClient) DeliverTxSync(tx []byte) (res types.Result) {
  246. reqres := cli.DeliverTxAsync(tx)
  247. if res := cli.checkErrGetResult(); res.IsErr() {
  248. return res
  249. }
  250. resp := reqres.Response.GetDeliverTx()
  251. return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
  252. }
  253. func (cli *grpcClient) CheckTxSync(tx []byte) (res types.Result) {
  254. reqres := cli.CheckTxAsync(tx)
  255. if res := cli.checkErrGetResult(); res.IsErr() {
  256. return res
  257. }
  258. resp := reqres.Response.GetCheckTx()
  259. return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
  260. }
  261. func (cli *grpcClient) QuerySync(query []byte) (res types.Result) {
  262. reqres := cli.QueryAsync(query)
  263. if res := cli.checkErrGetResult(); res.IsErr() {
  264. return res
  265. }
  266. resp := reqres.Response.GetQuery()
  267. return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
  268. }
  269. func (cli *grpcClient) CommitSync() (res types.Result) {
  270. reqres := cli.CommitAsync()
  271. if res := cli.checkErrGetResult(); res.IsErr() {
  272. return res
  273. }
  274. resp := reqres.Response.GetCommit()
  275. return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
  276. }
  277. func (cli *grpcClient) InitChainSync(validators []*types.Validator) (err error) {
  278. cli.InitChainAsync(validators)
  279. return cli.Error()
  280. }
  281. func (cli *grpcClient) BeginBlockSync(hash []byte, header *types.Header) (err error) {
  282. cli.BeginBlockAsync(hash, header)
  283. return cli.Error()
  284. }
  285. func (cli *grpcClient) EndBlockSync(height uint64) (resEndBlock types.ResponseEndBlock, err error) {
  286. reqres := cli.EndBlockAsync(height)
  287. if err := cli.Error(); err != nil {
  288. return resEndBlock, err
  289. }
  290. if resEndBlock_ := reqres.Response.GetEndBlock(); resEndBlock_ != nil {
  291. return *resEndBlock_, nil
  292. } else {
  293. return resEndBlock, nil
  294. }
  295. }