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.

156 lines
5.2 KiB

  1. package abciclient
  2. import (
  3. "context"
  4. "fmt"
  5. "sync"
  6. "github.com/tendermint/tendermint/abci/types"
  7. tmsync "github.com/tendermint/tendermint/internal/libs/sync"
  8. "github.com/tendermint/tendermint/libs/service"
  9. )
  10. const (
  11. dialRetryIntervalSeconds = 3
  12. echoRetryIntervalSeconds = 1
  13. )
  14. //go:generate ../../scripts/mockery_generate.sh Client
  15. // Client defines an interface for an ABCI client.
  16. //
  17. // All `Async` methods return a `ReqRes` object and an error.
  18. // All `Sync` methods return the appropriate protobuf ResponseXxx struct and an error.
  19. //
  20. // NOTE these are client errors, eg. ABCI socket connectivity issues.
  21. // Application-related errors are reflected in response via ABCI error codes
  22. // and logs.
  23. type Client interface {
  24. service.Service
  25. SetResponseCallback(Callback)
  26. Error() error
  27. // Asynchronous requests
  28. FlushAsync(context.Context) (*ReqRes, error)
  29. EchoAsync(ctx context.Context, msg string) (*ReqRes, error)
  30. InfoAsync(context.Context, types.RequestInfo) (*ReqRes, error)
  31. DeliverTxAsync(context.Context, types.RequestDeliverTx) (*ReqRes, error)
  32. CheckTxAsync(context.Context, types.RequestCheckTx) (*ReqRes, error)
  33. QueryAsync(context.Context, types.RequestQuery) (*ReqRes, error)
  34. CommitAsync(context.Context) (*ReqRes, error)
  35. InitChainAsync(context.Context, types.RequestInitChain) (*ReqRes, error)
  36. BeginBlockAsync(context.Context, types.RequestBeginBlock) (*ReqRes, error)
  37. EndBlockAsync(context.Context, types.RequestEndBlock) (*ReqRes, error)
  38. ListSnapshotsAsync(context.Context, types.RequestListSnapshots) (*ReqRes, error)
  39. OfferSnapshotAsync(context.Context, types.RequestOfferSnapshot) (*ReqRes, error)
  40. LoadSnapshotChunkAsync(context.Context, types.RequestLoadSnapshotChunk) (*ReqRes, error)
  41. ApplySnapshotChunkAsync(context.Context, types.RequestApplySnapshotChunk) (*ReqRes, error)
  42. // Synchronous requests
  43. FlushSync(context.Context) error
  44. EchoSync(ctx context.Context, msg string) (*types.ResponseEcho, error)
  45. InfoSync(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
  46. DeliverTxSync(context.Context, types.RequestDeliverTx) (*types.ResponseDeliverTx, error)
  47. CheckTxSync(context.Context, types.RequestCheckTx) (*types.ResponseCheckTx, error)
  48. QuerySync(context.Context, types.RequestQuery) (*types.ResponseQuery, error)
  49. CommitSync(context.Context) (*types.ResponseCommit, error)
  50. InitChainSync(context.Context, types.RequestInitChain) (*types.ResponseInitChain, error)
  51. BeginBlockSync(context.Context, types.RequestBeginBlock) (*types.ResponseBeginBlock, error)
  52. EndBlockSync(context.Context, types.RequestEndBlock) (*types.ResponseEndBlock, error)
  53. ListSnapshotsSync(context.Context, types.RequestListSnapshots) (*types.ResponseListSnapshots, error)
  54. OfferSnapshotSync(context.Context, types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
  55. LoadSnapshotChunkSync(context.Context, types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
  56. ApplySnapshotChunkSync(context.Context, types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
  57. }
  58. //----------------------------------------
  59. // NewClient returns a new ABCI client of the specified transport type.
  60. // It returns an error if the transport is not "socket" or "grpc"
  61. func NewClient(addr, transport string, mustConnect bool) (client Client, err error) {
  62. switch transport {
  63. case "socket":
  64. client = NewSocketClient(addr, mustConnect)
  65. case "grpc":
  66. client = NewGRPCClient(addr, mustConnect)
  67. default:
  68. err = fmt.Errorf("unknown abci transport %s", transport)
  69. }
  70. return
  71. }
  72. type Callback func(*types.Request, *types.Response)
  73. type ReqRes struct {
  74. *types.Request
  75. *sync.WaitGroup
  76. *types.Response // Not set atomically, so be sure to use WaitGroup.
  77. mtx tmsync.RWMutex
  78. done bool // Gets set to true once *after* WaitGroup.Done().
  79. cb func(*types.Response) // A single callback that may be set.
  80. }
  81. func NewReqRes(req *types.Request) *ReqRes {
  82. return &ReqRes{
  83. Request: req,
  84. WaitGroup: waitGroup1(),
  85. Response: nil,
  86. done: false,
  87. cb: nil,
  88. }
  89. }
  90. // Sets sets the callback. If reqRes is already done, it will call the cb
  91. // immediately. Note, reqRes.cb should not change if reqRes.done and only one
  92. // callback is supported.
  93. func (r *ReqRes) SetCallback(cb func(res *types.Response)) {
  94. r.mtx.Lock()
  95. if r.done {
  96. r.mtx.Unlock()
  97. cb(r.Response)
  98. return
  99. }
  100. r.cb = cb
  101. r.mtx.Unlock()
  102. }
  103. // InvokeCallback invokes a thread-safe execution of the configured callback
  104. // if non-nil.
  105. func (r *ReqRes) InvokeCallback() {
  106. r.mtx.Lock()
  107. defer r.mtx.Unlock()
  108. if r.cb != nil {
  109. r.cb(r.Response)
  110. }
  111. }
  112. // GetCallback returns the configured callback of the ReqRes object which may be
  113. // nil. Note, it is not safe to concurrently call this in cases where it is
  114. // marked done and SetCallback is called before calling GetCallback as that
  115. // will invoke the callback twice and create a potential race condition.
  116. //
  117. // ref: https://github.com/tendermint/tendermint/issues/5439
  118. func (r *ReqRes) GetCallback() func(*types.Response) {
  119. r.mtx.RLock()
  120. defer r.mtx.RUnlock()
  121. return r.cb
  122. }
  123. // SetDone marks the ReqRes object as done.
  124. func (r *ReqRes) SetDone() {
  125. r.mtx.Lock()
  126. defer r.mtx.Unlock()
  127. r.done = true
  128. }
  129. func waitGroup1() (wg *sync.WaitGroup) {
  130. wg = &sync.WaitGroup{}
  131. wg.Add(1)
  132. return
  133. }