mirror of
https://github.com/fluencelabs/tendermint
synced 2025-04-25 14:52:17 +00:00
62f97a69e9
As per #2127, this refactors the RequestCheckTx ProtoBuf struct to allow for a flag indicating whether a query is a recheck or not (and allows for possible future, more nuanced states). In order to pass this extended information through to the ABCI app, the proxy.AppConnMempool (and, for consistency, the proxy.AppConnConsensus) interface seems to need to be refactored along with abcicli.Client. And, as per this comment, I've made the following modification to the protobuf definition for the RequestCheckTx structure: enum CheckTxType { New = 0; Recheck = 1; } message RequestCheckTx { bytes tx = 1; CheckTxType type = 2; } * Refactor ABCI CheckTx to notify of recheck As per #2127, this refactors the `RequestCheckTx` ProtoBuf struct to allow for: 1. a flag indicating whether a query is a recheck or not (and allows for possible future, more nuanced states) 2. an `additional_data` bytes array to provide information for those more nuanced states. In order to pass this extended information through to the ABCI app, the `proxy.AppConnMempool` (and, for consistency, the `proxy.AppConnConsensus`) interface seems to need to be refactored. Commits: * Fix linting issue * Add CHANGELOG_PENDING entry * Remove extraneous explicit initialization * Update ABCI spec doc to include new CheckTx params * Rename method param for consistency * Rename CheckTxType enum values and remove additional_data param
130 lines
3.5 KiB
Go
130 lines
3.5 KiB
Go
package abcicli
|
|
|
|
import (
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/tendermint/tendermint/abci/types"
|
|
cmn "github.com/tendermint/tendermint/libs/common"
|
|
)
|
|
|
|
const (
|
|
dialRetryIntervalSeconds = 3
|
|
echoRetryIntervalSeconds = 1
|
|
)
|
|
|
|
// Client defines an interface for an ABCI client.
|
|
// All `Async` methods return a `ReqRes` object.
|
|
// All `Sync` methods return the appropriate protobuf ResponseXxx struct and an error.
|
|
// Note these are client errors, eg. ABCI socket connectivity issues.
|
|
// Application-related errors are reflected in response via ABCI error codes and logs.
|
|
type Client interface {
|
|
cmn.Service
|
|
|
|
SetResponseCallback(Callback)
|
|
Error() error
|
|
|
|
FlushAsync() *ReqRes
|
|
EchoAsync(msg string) *ReqRes
|
|
InfoAsync(types.RequestInfo) *ReqRes
|
|
SetOptionAsync(types.RequestSetOption) *ReqRes
|
|
DeliverTxAsync(types.RequestDeliverTx) *ReqRes
|
|
CheckTxAsync(types.RequestCheckTx) *ReqRes
|
|
QueryAsync(types.RequestQuery) *ReqRes
|
|
CommitAsync() *ReqRes
|
|
InitChainAsync(types.RequestInitChain) *ReqRes
|
|
BeginBlockAsync(types.RequestBeginBlock) *ReqRes
|
|
EndBlockAsync(types.RequestEndBlock) *ReqRes
|
|
|
|
FlushSync() error
|
|
EchoSync(msg string) (*types.ResponseEcho, error)
|
|
InfoSync(types.RequestInfo) (*types.ResponseInfo, error)
|
|
SetOptionSync(types.RequestSetOption) (*types.ResponseSetOption, error)
|
|
DeliverTxSync(types.RequestDeliverTx) (*types.ResponseDeliverTx, error)
|
|
CheckTxSync(types.RequestCheckTx) (*types.ResponseCheckTx, error)
|
|
QuerySync(types.RequestQuery) (*types.ResponseQuery, error)
|
|
CommitSync() (*types.ResponseCommit, error)
|
|
InitChainSync(types.RequestInitChain) (*types.ResponseInitChain, error)
|
|
BeginBlockSync(types.RequestBeginBlock) (*types.ResponseBeginBlock, error)
|
|
EndBlockSync(types.RequestEndBlock) (*types.ResponseEndBlock, error)
|
|
}
|
|
|
|
//----------------------------------------
|
|
|
|
// NewClient returns a new ABCI client of the specified transport type.
|
|
// It returns an error if the transport is not "socket" or "grpc"
|
|
func NewClient(addr, transport string, mustConnect bool) (client Client, err error) {
|
|
switch transport {
|
|
case "socket":
|
|
client = NewSocketClient(addr, mustConnect)
|
|
case "grpc":
|
|
client = NewGRPCClient(addr, mustConnect)
|
|
default:
|
|
err = fmt.Errorf("Unknown abci transport %s", transport)
|
|
}
|
|
return
|
|
}
|
|
|
|
//----------------------------------------
|
|
|
|
type Callback func(*types.Request, *types.Response)
|
|
|
|
//----------------------------------------
|
|
|
|
type ReqRes struct {
|
|
*types.Request
|
|
*sync.WaitGroup
|
|
*types.Response // Not set atomically, so be sure to use WaitGroup.
|
|
|
|
mtx sync.Mutex
|
|
done bool // Gets set to true once *after* WaitGroup.Done().
|
|
cb func(*types.Response) // A single callback that may be set.
|
|
}
|
|
|
|
func NewReqRes(req *types.Request) *ReqRes {
|
|
return &ReqRes{
|
|
Request: req,
|
|
WaitGroup: waitGroup1(),
|
|
Response: nil,
|
|
|
|
done: false,
|
|
cb: nil,
|
|
}
|
|
}
|
|
|
|
// Sets the callback for this ReqRes atomically.
|
|
// If reqRes is already done, calls cb immediately.
|
|
// NOTE: reqRes.cb should not change if reqRes.done.
|
|
// NOTE: only one callback is supported.
|
|
func (reqRes *ReqRes) SetCallback(cb func(res *types.Response)) {
|
|
reqRes.mtx.Lock()
|
|
|
|
if reqRes.done {
|
|
reqRes.mtx.Unlock()
|
|
cb(reqRes.Response)
|
|
return
|
|
}
|
|
|
|
reqRes.cb = cb
|
|
reqRes.mtx.Unlock()
|
|
}
|
|
|
|
func (reqRes *ReqRes) GetCallback() func(*types.Response) {
|
|
reqRes.mtx.Lock()
|
|
defer reqRes.mtx.Unlock()
|
|
return reqRes.cb
|
|
}
|
|
|
|
// NOTE: it should be safe to read reqRes.cb without locks after this.
|
|
func (reqRes *ReqRes) SetDone() {
|
|
reqRes.mtx.Lock()
|
|
reqRes.done = true
|
|
reqRes.mtx.Unlock()
|
|
}
|
|
|
|
func waitGroup1() (wg *sync.WaitGroup) {
|
|
wg = &sync.WaitGroup{}
|
|
wg.Add(1)
|
|
return
|
|
}
|