LaconicNetwork/cosmos-sdk
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

feat!: Comet v0.38 Integration (#15519)

Browse Source 
Co-authored-by: marbar3778 <marbar3778@yahoo.com>
Co-authored-by: cool-developer <51834436+cool-develope@users.noreply.github.com>
Co-authored-by: Aaron Craelius <aaron@regen.network>
Co-authored-by: Matt Kocubinski <mkocubinski@gmail.com>
Co-authored-by: Julien Robert <julien@rbrt.fr>
This commit is contained in:
Aleksandr Bezobchuk
2023-05-24 16:09:19 +00:00
committed by GitHub
co-authored by marbar3778 cool-developer Aaron Craelius Matt Kocubinski Julien Robert
parent 166be3766b
commit 6cee22df52
148 changed files with 14747 additions and 15262 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/go.mod
+2 -2
View File
@@ -7,7 +7,7 @@ require (
cosmossdk.io/log v1.1.0
cosmossdk.io/math v1.0.1
github.com/armon/go-metrics v0.4.1
github.com/cometbft/cometbft v0.37.1
github.com/cometbft/cometbft v0.38.0-alpha.1
github.com/cosmos/cosmos-db v1.0.0-rc.1
github.com/cosmos/gogoproto v1.4.10
github.com/cosmos/iavl v0.21.0
@@ -39,7 +39,6 @@ require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
github.com/fatih/color v1.15.0 // indirect
github.com/fsnotify/fsnotify v1.6.0 // indirect
github.com/getsentry/sentry-go v0.18.0 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/snappy v0.0.4 // indirect
@@ -57,6 +56,7 @@ require (
github.com/mattn/go-isatty v0.0.19 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
github.com/mitchellh/go-testing-interface v1.14.1 // indirect
github.com/oasisprotocol/curve25519-voi v0.0.0-20220708102147-0a8a51822cae // indirect
github.com/oklog/run v1.1.0 // indirect
github.com/onsi/gomega v1.20.0 // indirect
github.com/petermattis/goid v0.0.0-20221215004737-a150e88a970d // indirect
store/go.sum
+6 -6
View File
@@ -29,9 +29,9 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/btcsuite/btcd/btcec/v2 v2.3.2 h1:5n0X6hX0Zk+6omWcihdYvdAlGf2DfasC0GMf7DClJ3U=
github.com/btcsuite/btcd/btcec/v2 v2.3.2/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04=
github.com/btcsuite/btcd/btcutil v1.1.2 h1:XLMbX8JQEiwMcYft2EGi8zPUkoa0abKIU6/BJSRsjzQ=
github.com/btcsuite/btcd/btcutil v1.1.3 h1:xfbtw8lwpp0G6NwSHb+UE67ryTFHJAiNuipusjXSohQ=
github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 h1:q0rUy8C/TYNBQS1+CGKw68tLOFYSNEs0TFnxxnS9+4U=
github.com/bufbuild/protocompile v0.4.0 h1:LbFKd2XowZvQ/kajzguUp2DC9UEIQhIq77fZZlaQsNA=
github.com/bufbuild/protocompile v0.5.1 h1:mixz5lJX4Hiz4FpqFREJHIXLfaLBntfaJv1h+/jS+Qg=
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
@@ -55,8 +55,8 @@ github.com/cockroachdb/pebble v0.0.0-20230226194802-02d779ffbc46/go.mod h1:9lRMC
github.com/cockroachdb/redact v1.1.3 h1:AKZds10rFSIj7qADf0g46UixK8NNLwWTNdCIGS5wfSQ=
github.com/cockroachdb/redact v1.1.3/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg=
github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0/go.mod h1:4Zcjuz89kmFXt9morQgcfYZAYZ5n8WHjt81YYWIwtTM=
github.com/cometbft/cometbft v0.37.1 h1:KLxkQTK2hICXYq21U2hn1W5hOVYUdQgDQ1uB+90xPIg=
github.com/cometbft/cometbft v0.37.1/go.mod h1:Y2MMMN//O5K4YKd8ze4r9jmk4Y7h0ajqILXbH5JQFVs=
github.com/cometbft/cometbft v0.38.0-alpha.1 h1:BmDIGG49GeSpx4BXtyGqIfTkTUDXFHgvfaFNGETGH0o=
github.com/cometbft/cometbft v0.38.0-alpha.1/go.mod h1:5Jz0Z8YsHSf0ZaAqGvi/ifioSdVFPtEGrm8Y9T/993k=
github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -96,7 +96,6 @@ github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMo
github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
github.com/gavv/httpexpect v2.0.0+incompatible/go.mod h1:x+9tiU1YnrOvnB725RkpoLv1M62hOWzwo5OXotisrKc=
github.com/getsentry/sentry-go v0.12.0/go.mod h1:NSap0JBYWzHND8oMbyi0+XZhUalc1TBdRL1M71JZW2c=
github.com/getsentry/sentry-go v0.18.0 h1:MtBW5H9QgdcJabtZcuJG80BMOwaBpkRDZkxRkNC1sN0=
@@ -263,6 +262,8 @@ github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OS
github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
github.com/oasisprotocol/curve25519-voi v0.0.0-20220708102147-0a8a51822cae h1:FatpGJD2jmJfhZiFDElaC0QhZUDQnxUeAwTGkfAHN3I=
github.com/oasisprotocol/curve25519-voi v0.0.0-20220708102147-0a8a51822cae/go.mod h1:hVoHR2EVESiICEMbg137etN/Lx+lSrHPTD39Z/uE+2s=
github.com/oklog/run v1.1.0 h1:GEenZ1cK0+q0+wsJew9qUg/DyD8k3JzYsZAi5gYi2mA=
github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DVU=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@@ -470,7 +471,6 @@ golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
store/iavl/store.go
+5 -3
View File
@@ -304,7 +304,7 @@ func (st *Store) Import(version int64) (*iavl.Importer, error) {
}
// Handle gatest the latest height, if height is 0
func getHeight(tree Tree, req abci.RequestQuery) int64 {
func getHeight(tree Tree, req *abci.RequestQuery) int64 {
height := req.Height
if height == 0 {
latest := tree.Version()
@@ -324,7 +324,7 @@ func getHeight(tree Tree, req abci.RequestQuery) int64 {
// If latest-1 is not present, use latest (which must be present)
// if you care to have the latest data to see a tx results, you must
// explicitly set the height you want to see
func (st *Store) Query(req abci.RequestQuery) (res abci.ResponseQuery) {
func (st *Store) Query(req *abci.RequestQuery) (res *abci.ResponseQuery) {
defer st.metrics.MeasureSince("store", "iavl", "query")
if len(req.Data) == 0 {
@@ -335,7 +335,9 @@ func (st *Store) Query(req abci.RequestQuery) (res abci.ResponseQuery) {
// store the height we chose in the response, with 0 being changed to the
// latest height
res.Height = getHeight(tree, req)
res = &abci.ResponseQuery{
Height: getHeight(tree, req),
}
switch req.Path {
case "/key": // get by key
store/iavl/store_test.go
+11 -11
View File
@@ -138,7 +138,7 @@ func TestGetImmutable(t *testing.T) {
require.NoError(t, err)
require.Equal(t, newStore.Get([]byte("hello")), []byte("adios"))
res := newStore.Query(abci.RequestQuery{Data: []byte("hello"), Height: cID.Version, Path: "/key", Prove: true})
res := newStore.Query(&abci.RequestQuery{Data: []byte("hello"), Height: cID.Version, Path: "/key", Prove: true})
require.Equal(t, res.Value, []byte("adios"))
require.NotNil(t, res.ProofOps)
@@ -505,7 +505,7 @@ func TestIAVLStoreQuery(t *testing.T) {
querySub := abci.RequestQuery{Path: "/subspace", Data: ksub, Height: ver}
// query subspace before anything set
qres := iavlStore.Query(querySub)
qres := iavlStore.Query(&querySub)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, valExpSubEmpty, qres.Value)
@@ -514,24 +514,24 @@ func TestIAVLStoreQuery(t *testing.T) {
iavlStore.Set(k2, v2)
// set data without commit, doesn't show up
qres = iavlStore.Query(query)
qres = iavlStore.Query(&query)
require.Equal(t, uint32(0), qres.Code)
require.Nil(t, qres.Value)
// commit it, but still don't see on old version
cid = iavlStore.Commit()
qres = iavlStore.Query(query)
qres = iavlStore.Query(&query)
require.Equal(t, uint32(0), qres.Code)
require.Nil(t, qres.Value)
// but yes on the new version
query.Height = cid.Version
qres = iavlStore.Query(query)
qres = iavlStore.Query(&query)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, v1, qres.Value)
// and for the subspace
qres = iavlStore.Query(querySub)
qres = iavlStore.Query(&querySub)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, valExpSub1, qres.Value)
@@ -540,28 +540,28 @@ func TestIAVLStoreQuery(t *testing.T) {
cid = iavlStore.Commit()
// query will return old values, as height is fixed
qres = iavlStore.Query(query)
qres = iavlStore.Query(&query)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, v1, qres.Value)
// update to latest in the query and we are happy
query.Height = cid.Version
qres = iavlStore.Query(query)
qres = iavlStore.Query(&query)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, v3, qres.Value)
query2 := abci.RequestQuery{Path: "/key", Data: k2, Height: cid.Version}
qres = iavlStore.Query(query2)
qres = iavlStore.Query(&query2)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, v2, qres.Value)
// and for the subspace
qres = iavlStore.Query(querySub)
qres = iavlStore.Query(&querySub)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, valExpSub2, qres.Value)
// default (height 0) will show latest -1
query0 := abci.RequestQuery{Path: "/key", Data: k1}
qres = iavlStore.Query(query0)
qres = iavlStore.Query(&query0)
require.Equal(t, uint32(0), qres.Code)
require.Equal(t, v1, qres.Value)
}
store/rootmulti/proof_test.go
+3 -3
View File
@@ -23,7 +23,7 @@ func TestVerifyIAVLStoreQueryProof(t *testing.T) {
cid := store.Commit()
// Get Proof
res := store.Query(abci.RequestQuery{
res := store.Query(&abci.RequestQuery{
Path: "/key", // required path to get key/value+proof
Data: []byte("MYKEY"),
Prove: true,
@@ -70,7 +70,7 @@ func TestVerifyMultiStoreQueryProof(t *testing.T) {
cid := store.Commit()
// Get Proof
res := store.Query(abci.RequestQuery{
res := store.Query(&abci.RequestQuery{
Path: "/iavlStoreKey/key", // required path to get key/value+proof
Data: []byte("MYKEY"),
Prove: true,
@@ -126,7 +126,7 @@ func TestVerifyMultiStoreQueryProofAbsence(t *testing.T) {
cid := store.Commit() // Commit with empty iavl store.
// Get Proof
res := store.Query(abci.RequestQuery{
res := store.Query(&abci.RequestQuery{
Path: "/iavlStoreKey/key", // required path to get key/value+proof
Data: []byte("MYABSENTKEY"),
Prove: true,
store/rootmulti/store.go
+1 -1
View File
@@ -726,7 +726,7 @@ func (rs *Store) GetStoreByName(name string) types.Store {
// modified to remove the substore prefix.
// Ie. `req.Path` here is `/<substore>/<path>`, and trimmed to `/<path>` for the substore.
// TODO: add proof for `multistore -> substore`.
func (rs *Store) Query(req abci.RequestQuery) abci.ResponseQuery {
func (rs *Store) Query(req *abci.RequestQuery) *abci.ResponseQuery {
path := req.Path
storeName, subpath, err := parsePath(path)
if err != nil {
store/rootmulti/store_test.go
+8 -6
View File
@@ -457,37 +457,39 @@ func TestMultiStoreQuery(t *testing.T) {
// Test bad path.
query := abci.RequestQuery{Path: "/key", Data: k, Height: ver}
qres := multi.Query(query)
qres := multi.Query(&query)
require.EqualValues(t, types.ErrUnknownRequest.ABCICode(), qres.Code)
require.EqualValues(t, types.ErrUnknownRequest.Codespace(), qres.Codespace)
query.Path = "h897fy32890rf63296r92"
qres = multi.Query(query)
qres = multi.Query(&query)
require.EqualValues(t, types.ErrUnknownRequest.ABCICode(), qres.Code)
require.EqualValues(t, types.ErrUnknownRequest.Codespace(), qres.Codespace)
// Test invalid store name.
query.Path = "/garbage/key"
qres = multi.Query(query)
qres = multi.Query(&query)
require.EqualValues(t, types.ErrUnknownRequest.ABCICode(), qres.Code)
require.EqualValues(t, types.ErrUnknownRequest.Codespace(), qres.Codespace)
// Test valid query with data.
query.Path = "/store1/key"
qres = multi.Query(query)
qres = multi.Query(&query)
require.EqualValues(t, 0, qres.Code)
require.Equal(t, v, qres.Value)
// Test valid but empty query.
query.Path = "/store2/key"
query.Prove = true
qres = multi.Query(query)
qres = multi.Query(&query)
require.EqualValues(t, 0, qres.Code)
require.Nil(t, qres.Value)
// Test store2 data.
// Since we are using the request as a reference, the path will be modified.
query.Data = k2
qres = multi.Query(query)
query.Path = "/store2/key"
qres = multi.Query(&query)
require.EqualValues(t, 0, qres.Code)
require.Equal(t, v2, qres.Value)
}
store/streaming/abci/examples/file/file.go
+3 -28
View File
@@ -43,38 +43,13 @@ func (a *FilePlugin) writeToFile(file string, data []byte) error {
return nil
}
func (a *FilePlugin) ListenBeginBlock(ctx context.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error {
a.BlockHeight = req.Header.Height
d1 := []byte(fmt.Sprintf("%d:::%v\n", a.BlockHeight, req))
d2 := []byte(fmt.Sprintf("%d:::%v\n", a.BlockHeight, res))
if err := a.writeToFile("begin-block-req", d1); err != nil {
return err
}
if err := a.writeToFile("begin-block-res", d2); err != nil {
return err
}
return nil
}
func (a *FilePlugin) ListenEndBlock(ctx context.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error {
func (a *FilePlugin) ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock) error {
d1 := []byte(fmt.Sprintf("%d:::%v\n", a.BlockHeight, req))
d2 := []byte(fmt.Sprintf("%d:::%v\n", a.BlockHeight, req))
if err := a.writeToFile("end-block-req", d1); err != nil {
if err := a.writeToFile("finalize-block-req", d1); err != nil {
return err
}
if err := a.writeToFile("end-block-res", d2); err != nil {
return err
}
return nil
}
func (a *FilePlugin) ListenDeliverTx(ctx context.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error {
d1 := []byte(fmt.Sprintf("%d:::%v\n", a.BlockHeight, req))
d2 := []byte(fmt.Sprintf("%d:::%v\n", a.BlockHeight, res))
if err := a.writeToFile("deliver-tx-req", d1); err != nil {
return err
}
if err := a.writeToFile("deliver-tx-res", d2); err != nil {
if err := a.writeToFile("finalize-block-res", d2); err != nil {
return err
}
return nil
store/streaming/abci/examples/stdout/stdout
BIN
View File
Binary file not shown.
store/streaming/abci/examples/stdout/stdout.go
+3 -15
View File
@@ -17,22 +17,10 @@ type StdoutPlugin struct {
BlockHeight int64
}
func (a *StdoutPlugin) ListenBeginBlock(ctx context.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error {
a.BlockHeight = req.Header.Height
func (a *StdoutPlugin) ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock) error {
a.BlockHeight = req.Height
// process tx messages (i.e: sent to external system)
fmt.Printf("listen-begin-block: block-height=%d req=%v res=%v", a.BlockHeight, req, res)
return nil
}
func (a *StdoutPlugin) ListenEndBlock(ctx context.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error {
// process end block messages (i.e: sent to external system)
fmt.Printf("listen-end-block: block-height=%d req=%v res=%v", a.BlockHeight, req, res)
return nil
}
func (a *StdoutPlugin) ListenDeliverTx(ctx context.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error {
// process tx messages (i.e: sent to external system)
fmt.Printf("listen-deliver-tx: block-height=%d req=%v res=%v", a.BlockHeight, req, res)
fmt.Printf("listen-finalize-block: block-height=%d req=%v res=%v", a.BlockHeight, req, res)
return nil
}
store/streaming/abci/grpc.go
+7 -55
View File
@@ -17,52 +17,18 @@ type GRPCClient struct {
client ABCIListenerServiceClient
}
// ListenBeginBlock listens to begin block request and responses.
// In addition, it retrieves a types.Context from a context.Context instance.
// It panics if a types.Context was not properly attached.
// When the node is configured to stop on listening errors,
// it will terminate immediately and exit with a non-zero code.
func (m *GRPCClient) ListenBeginBlock(goCtx context.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error {
ctx := goCtx.(storetypes.Context)
sm := ctx.StreamingManager()
request := &ListenBeginBlockRequest{Req: &req, Res: &res}
_, err := m.client.ListenBeginBlock(goCtx, request)
if err != nil && sm.StopNodeOnErr {
ctx.Logger().Error("BeginBlock listening hook failed", "height", ctx.BlockHeight(), "err", err)
cleanupAndExit()
}
return err
}
// ListenEndBlock listens to end block request and responses.
// In addition, it retrieves a types.Context from a context.Context instance.
// It panics if a types.Context was not properly attached.
// When the node is configured to stop on listening errors,
// it will terminate immediately and exit with a non-zero code.
func (m *GRPCClient) ListenEndBlock(goCtx context.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error {
func (m *GRPCClient) ListenFinalizeBlock(goCtx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock) error {
ctx := goCtx.(storetypes.Context)
sm := ctx.StreamingManager()
request := &ListenEndBlockRequest{Req: &req, Res: &res}
_, err := m.client.ListenEndBlock(goCtx, request)
request := &ListenFinalizeBlockRequest{Req: &req, Res: &res}
_, err := m.client.ListenFinalizeBlock(goCtx, request)
if err != nil && sm.StopNodeOnErr {
ctx.Logger().Error("EndBlock listening hook failed", "height", ctx.BlockHeight(), "err", err)
cleanupAndExit()
}
return err
}
// ListenDeliverTx listens to deliver tx request and responses.
// In addition, it retrieves a types.Context from a context.Context instance.
// It panics if a types.Context was not properly attached.
// When the node is configured to stop on listening errors,
// it will terminate immediately and exit with a non-zero code.
func (m *GRPCClient) ListenDeliverTx(goCtx context.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error {
ctx := goCtx.(storetypes.Context)
sm := ctx.StreamingManager()
request := &ListenDeliverTxRequest{BlockHeight: ctx.BlockHeight(), Req: &req, Res: &res}
_, err := m.client.ListenDeliverTx(goCtx, request)
if err != nil && sm.StopNodeOnErr {
ctx.Logger().Error("DeliverTx listening hook failed", "height", ctx.BlockHeight(), "err", err)
ctx.Logger().Error("FinalizeBlock listening hook failed", "height", ctx.BlockHeight(), "err", err)
cleanupAndExit()
}
return err
@@ -98,25 +64,11 @@ type GRPCServer struct {
Impl storetypes.ABCIListener
}
func (m GRPCServer) ListenBeginBlock(ctx context.Context, request *ListenBeginBlockRequest) (*ListenBeginBlockResponse, error) {
if err := m.Impl.ListenBeginBlock(ctx, *request.Req, *request.Res); err != nil {
func (m GRPCServer) ListenFinalizeBlock(ctx context.Context, request *ListenFinalizeBlockRequest) (*ListenFinalizeBlockResponse, error) {
if err := m.Impl.ListenFinalizeBlock(ctx, *request.Req, *request.Res); err != nil {
return nil, err
}
return &ListenBeginBlockResponse{}, nil
}
func (m GRPCServer) ListenEndBlock(ctx context.Context, request *ListenEndBlockRequest) (*ListenEndBlockResponse, error) {
if err := m.Impl.ListenEndBlock(ctx, *request.Req, *request.Res); err != nil {
return nil, err
}
return &ListenEndBlockResponse{}, nil
}
func (m GRPCServer) ListenDeliverTx(ctx context.Context, request *ListenDeliverTxRequest) (*ListenDeliverTxResponse, error) {
if err := m.Impl.ListenDeliverTx(ctx, *request.Req, *request.Res); err != nil {
return nil, err
}
return &ListenDeliverTxResponse{}, nil
return &ListenFinalizeBlockResponse{}, nil
}
func (m GRPCServer) ListenCommit(ctx context.Context, request *ListenCommitRequest) (*ListenCommitResponse, error) {
store/streaming/abci/grpc.pb.go
+97 -936
View File
File diff suppressed because it is too large Load Diff
store/streaming/streaming_test.go
+22 -38
View File
@@ -27,13 +27,9 @@ type PluginTestSuite struct {
workDir string
beginBlockReq abci.RequestBeginBlock
beginBlockRes abci.ResponseBeginBlock
endBlockReq abci.RequestEndBlock
endBlockRes abci.ResponseEndBlock
deliverTxReq abci.RequestDeliverTx
deliverTxRes abci.ResponseDeliverTx
commitRes abci.ResponseCommit
finalizeBlockReq abci.RequestFinalizeBlock
finalizeBlockRes abci.ResponseFinalizeBlock
commitRes abci.ResponseCommit
changeSet []*storetypes.StoreKVPair
}
@@ -71,33 +67,28 @@ func (s *PluginTestSuite) SetupTest() {
s.loggerCtx = NewMockContext(header, logger, streamingService)
// test abci message types
s.beginBlockReq = abci.RequestBeginBlock{
Header: s.loggerCtx.BlockHeader(),
ByzantineValidators: []abci.Misbehavior{},
Hash: []byte{1, 2, 3, 4, 5, 6, 7, 8, 9},
LastCommitInfo: abci.CommitInfo{Round: 1, Votes: []abci.VoteInfo{}},
s.finalizeBlockReq = abci.RequestFinalizeBlock{
Height: s.loggerCtx.BlockHeight(),
Txs: [][]byte{{1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}},
Misbehavior: []abci.Misbehavior{},
Hash: []byte{1, 2, 3, 4, 5, 6, 7, 8, 9},
DecidedLastCommit: abci.CommitInfo{},
}
s.beginBlockRes = abci.ResponseBeginBlock{
Events: []abci.Event{{Type: "testEventType1"}},
}
s.endBlockReq = abci.RequestEndBlock{Height: s.loggerCtx.BlockHeight()}
s.endBlockRes = abci.ResponseEndBlock{
s.finalizeBlockRes = abci.ResponseFinalizeBlock{
Events: []abci.Event{},
ConsensusParamUpdates: &tmproto.ConsensusParams{},
ValidatorUpdates: []abci.ValidatorUpdate{},
}
s.deliverTxReq = abci.RequestDeliverTx{
Tx: []byte{9, 8, 7, 6, 5, 4, 3, 2, 1},
}
s.deliverTxRes = abci.ResponseDeliverTx{
Events: []abci.Event{},
Code: 1,
Codespace: "mockCodeSpace",
Data: []byte{5, 6, 7, 8},
GasUsed: 2,
GasWanted: 3,
Info: "mockInfo",
Log: "mockLog",
TxResults: []*abci.ExecTxResult{{
Events: []abci.Event{},
Code: 1,
Codespace: "mockCodeSpace",
Data: []byte{5, 6, 7, 8},
GasUsed: 2,
GasWanted: 3,
Info: "mockInfo",
Log: "mockLog",
}},
}
s.commitRes = abci.ResponseCommit{}
@@ -121,17 +112,10 @@ func (s *PluginTestSuite) TestABCIGRPCPlugin() {
abciListeners := s.loggerCtx.StreamingManager().ABCIListeners
for _, abciListener := range abciListeners {
for i := range [50]int{} {
err := abciListener.ListenBeginBlock(s.loggerCtx, s.beginBlockReq, s.beginBlockRes)
assert.NoError(t, err, "ListenBeginBlock")
err = abciListener.ListenEndBlock(s.loggerCtx, s.endBlockReq, s.endBlockRes)
err := abciListener.ListenFinalizeBlock(s.loggerCtx, s.finalizeBlockReq, s.finalizeBlockRes)
assert.NoError(t, err, "ListenEndBlock")
for range [50]int{} {
err = abciListener.ListenDeliverTx(s.loggerCtx, s.deliverTxReq, s.deliverTxRes)
assert.NoError(t, err, "ListenDeliverTx")
}
err = abciListener.ListenCommit(s.loggerCtx, s.commitRes, s.changeSet)
assert.NoError(t, err, "ListenCommit")
store/types/errors.go
+2 -2
View File
@@ -32,9 +32,9 @@ var (
// QueryResult returns a ResponseQuery from an error. It will try to parse ABCI
// info from the error.
func QueryResult(err error, debug bool) abci.ResponseQuery {
func QueryResult(err error, debug bool) *abci.ResponseQuery {
space, code, log := errors.ABCIInfo(err, debug)
return abci.ResponseQuery{
return &abci.ResponseQuery{
Codespace: space,
Code: code,
Log: log,
store/types/listening.pb.go
+70 -497
View File
@@ -99,12 +99,9 @@ func (m *StoreKVPair) GetValue() []byte {
// BlockMetadata contains all the abci event data of a block
// the file streamer dump them into files together with the state changes.
type BlockMetadata struct {
RequestBeginBlock *types.RequestBeginBlock `protobuf:"bytes,1,opt,name=request_begin_block,json=requestBeginBlock,proto3" json:"request_begin_block,omitempty"`
ResponseBeginBlock *types.ResponseBeginBlock `protobuf:"bytes,2,opt,name=response_begin_block,json=responseBeginBlock,proto3" json:"response_begin_block,omitempty"`
DeliverTxs []*BlockMetadata_DeliverTx `protobuf:"bytes,3,rep,name=deliver_txs,json=deliverTxs,proto3" json:"deliver_txs,omitempty"`
RequestEndBlock *types.RequestEndBlock `protobuf:"bytes,4,opt,name=request_end_block,json=requestEndBlock,proto3" json:"request_end_block,omitempty"`
ResponseEndBlock *types.ResponseEndBlock `protobuf:"bytes,5,opt,name=response_end_block,json=responseEndBlock,proto3" json:"response_end_block,omitempty"`
ResponseCommit *types.ResponseCommit `protobuf:"bytes,6,opt,name=response_commit,json=responseCommit,proto3" json:"response_commit,omitempty"`
ResponseCommit *types.ResponseCommit `protobuf:"bytes,6,opt,name=response_commit,json=responseCommit,proto3" json:"response_commit,omitempty"`
RequestFinalizeBlock *types.RequestFinalizeBlock `protobuf:"bytes,7,opt,name=request_finalize_block,json=requestFinalizeBlock,proto3" json:"request_finalize_block,omitempty"`
ResponseFinalizeBlock *types.ResponseFinalizeBlock `protobuf:"bytes,8,opt,name=response_finalize_block,json=responseFinalizeBlock,proto3" json:"response_finalize_block,omitempty"`
}
func (m *BlockMetadata) Reset() { *m = BlockMetadata{} }
@@ -140,41 +137,6 @@ func (m *BlockMetadata) XXX_DiscardUnknown() {
var xxx_messageInfo_BlockMetadata proto.InternalMessageInfo
func (m *BlockMetadata) GetRequestBeginBlock() *types.RequestBeginBlock {
if m != nil {
return m.RequestBeginBlock
}
return nil
}
func (m *BlockMetadata) GetResponseBeginBlock() *types.ResponseBeginBlock {
if m != nil {
return m.ResponseBeginBlock
}
return nil
}
func (m *BlockMetadata) GetDeliverTxs() []*BlockMetadata_DeliverTx {
if m != nil {
return m.DeliverTxs
}
return nil
}
func (m *BlockMetadata) GetRequestEndBlock() *types.RequestEndBlock {
if m != nil {
return m.RequestEndBlock
}
return nil
}
func (m *BlockMetadata) GetResponseEndBlock() *types.ResponseEndBlock {
if m != nil {
return m.ResponseEndBlock
}
return nil
}
func (m *BlockMetadata) GetResponseCommit() *types.ResponseCommit {
if m != nil {
return m.ResponseCommit
@@ -182,55 +144,16 @@ func (m *BlockMetadata) GetResponseCommit() *types.ResponseCommit {
return nil
}
// DeliverTx encapulate deliver tx request and response.
type BlockMetadata_DeliverTx struct {
Request *types.RequestDeliverTx `protobuf:"bytes,1,opt,name=request,proto3" json:"request,omitempty"`
Response *types.ResponseDeliverTx `protobuf:"bytes,2,opt,name=response,proto3" json:"response,omitempty"`
}
func (m *BlockMetadata_DeliverTx) Reset() { *m = BlockMetadata_DeliverTx{} }
func (m *BlockMetadata_DeliverTx) String() string { return proto.CompactTextString(m) }
func (*BlockMetadata_DeliverTx) ProtoMessage() {}
func (*BlockMetadata_DeliverTx) Descriptor() ([]byte, []int) {
return fileDescriptor_b6caeb9d7b7c7c10, []int{1, 0}
}
func (m *BlockMetadata_DeliverTx) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *BlockMetadata_DeliverTx) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_BlockMetadata_DeliverTx.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *BlockMetadata_DeliverTx) XXX_Merge(src proto.Message) {
xxx_messageInfo_BlockMetadata_DeliverTx.Merge(m, src)
}
func (m *BlockMetadata_DeliverTx) XXX_Size() int {
return m.Size()
}
func (m *BlockMetadata_DeliverTx) XXX_DiscardUnknown() {
xxx_messageInfo_BlockMetadata_DeliverTx.DiscardUnknown(m)
}
var xxx_messageInfo_BlockMetadata_DeliverTx proto.InternalMessageInfo
func (m *BlockMetadata_DeliverTx) GetRequest() *types.RequestDeliverTx {
func (m *BlockMetadata) GetRequestFinalizeBlock() *types.RequestFinalizeBlock {
if m != nil {
return m.Request
return m.RequestFinalizeBlock
}
return nil
}
func (m *BlockMetadata_DeliverTx) GetResponse() *types.ResponseDeliverTx {
func (m *BlockMetadata) GetResponseFinalizeBlock() *types.ResponseFinalizeBlock {
if m != nil {
return m.Response
return m.ResponseFinalizeBlock
}
return nil
}
@@ -238,7 +161,6 @@ func (m *BlockMetadata_DeliverTx) GetResponse() *types.ResponseDeliverTx {
func init() {
proto.RegisterType((*StoreKVPair)(nil), "cosmos.store.v1beta1.StoreKVPair")
proto.RegisterType((*BlockMetadata)(nil), "cosmos.store.v1beta1.BlockMetadata")
proto.RegisterType((*BlockMetadata_DeliverTx)(nil), "cosmos.store.v1beta1.BlockMetadata.DeliverTx")
}
func init() {
@@ -246,36 +168,31 @@ func init() {
}
var fileDescriptor_b6caeb9d7b7c7c10 = []byte{
// 458 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x93, 0x4f, 0x6f, 0xd3, 0x30,
0x18, 0xc6, 0x9b, 0xb5, 0x2b, 0xed, 0x5b, 0x60, 0xc3, 0x54, 0x28, 0xda, 0xa4, 0x90, 0x15, 0x0e,
0xbd, 0xe0, 0x68, 0xe5, 0x88, 0xc4, 0xa1, 0x80, 0x84, 0x34, 0xfe, 0x29, 0xfc, 0x39, 0x70, 0x89,
0x92, 0xe6, 0xd5, 0x64, 0x9a, 0xda, 0xc5, 0xf6, 0xaa, 0xf5, 0x1b, 0x70, 0xe4, 0x63, 0x71, 0xdc,
0x91, 0x23, 0x6a, 0xbf, 0x07, 0x42, 0xb1, 0xd3, 0x74, 0x19, 0xcd, 0xcd, 0x7e, 0xf2, 0x3c, 0xbf,
0x3c, 0xaf, 0x2d, 0xc3, 0xe3, 0x89, 0x50, 0x33, 0xa1, 0x02, 0xa5, 0x85, 0xc4, 0x60, 0x71, 0x9a,
0xa0, 0x8e, 0x4f, 0x83, 0x8c, 0x29, 0x8d, 0x9c, 0xf1, 0x73, 0x3a, 0x97, 0x42, 0x0b, 0xd2, 0xb7,
0x2e, 0x6a, 0x5c, 0xb4, 0x70, 0x1d, 0x1d, 0x6b, 0xe4, 0x29, 0xca, 0x19, 0xe3, 0x3a, 0x88, 0x93,
0x09, 0x0b, 0xf4, 0x72, 0x8e, 0xca, 0x46, 0x06, 0xdf, 0xa0, 0xf7, 0x31, 0x77, 0x9f, 0x7d, 0xf9,
0x10, 0x33, 0x49, 0x8e, 0xa1, 0x6b, 0xc2, 0xd1, 0x14, 0x97, 0xae, 0xe3, 0x3b, 0xc3, 0x6e, 0xd8,
0x31, 0xc2, 0x19, 0x2e, 0xc9, 0x03, 0x68, 0xa7, 0x98, 0xa1, 0x46, 0x77, 0xcf, 0x77, 0x86, 0x9d,
0xb0, 0xd8, 0x91, 0x43, 0x68, 0xe6, 0xf6, 0xa6, 0xef, 0x0c, 0x6f, 0x87, 0xf9, 0x92, 0xf4, 0x61,
0x7f, 0x11, 0x67, 0x17, 0xe8, 0xb6, 0x8c, 0x66, 0x37, 0x83, 0xbf, 0x2d, 0xb8, 0x33, 0xce, 0xc4,
0x64, 0xfa, 0x16, 0x75, 0x9c, 0xc6, 0x3a, 0x26, 0x21, 0xdc, 0x97, 0xf8, 0xfd, 0x02, 0x95, 0x8e,
0x12, 0x3c, 0x67, 0x3c, 0x4a, 0xf2, 0xcf, 0xe6, 0xc7, 0xbd, 0xd1, 0x80, 0x6e, 0x8b, 0xd3, 0xbc,
0x38, 0x0d, 0xad, 0x77, 0x9c, 0x5b, 0x0d, 0x28, 0xbc, 0x27, 0x6f, 0x4a, 0xe4, 0x33, 0xf4, 0x25,
0xaa, 0xb9, 0xe0, 0x0a, 0x2b, 0xd0, 0x3d, 0x03, 0x7d, 0xb4, 0x03, 0x6a, 0xcd, 0xd7, 0xa8, 0x44,
0xfe, 0xa7, 0x91, 0x77, 0xd0, 0x4b, 0x31, 0x63, 0x0b, 0x94, 0x91, 0xbe, 0x54, 0x6e, 0xd3, 0x6f,
0x0e, 0x7b, 0xa3, 0x27, 0x74, 0xd7, 0x89, 0xd3, 0xca, 0x90, 0xf4, 0xa5, 0x8d, 0x7d, 0xba, 0x0c,
0x21, 0xdd, 0x2c, 0x15, 0x79, 0x03, 0x9b, 0xee, 0x11, 0xf2, 0xb4, 0xe8, 0xd8, 0x32, 0x1d, 0xfd,
0xba, 0xc1, 0x5f, 0xf1, 0xd4, 0x16, 0x3c, 0x90, 0x55, 0x81, 0xbc, 0x87, 0xb2, 0xf3, 0x35, 0xdc,
0xbe, 0xc1, 0x9d, 0xd4, 0x8e, 0x5c, 0xf2, 0x0e, 0xe5, 0x0d, 0x85, 0xbc, 0x86, 0x83, 0x12, 0x38,
0x11, 0xb3, 0x19, 0xd3, 0x6e, 0xdb, 0xd0, 0x1e, 0xd6, 0xd2, 0x5e, 0x18, 0x5b, 0x78, 0x57, 0x56,
0xf6, 0x47, 0x3f, 0x1c, 0xe8, 0x96, 0x47, 0x40, 0x9e, 0xc1, 0xad, 0xa2, 0x7b, 0x71, 0xcb, 0x27,
0x75, 0xc3, 0x6e, 0x8f, 0x6d, 0x93, 0x20, 0xcf, 0xa1, 0xb3, 0x81, 0x17, 0xd7, 0x39, 0xa8, 0x6d,
0xb3, 0x8d, 0x97, 0x99, 0xf1, 0xe8, 0xd7, 0xca, 0x73, 0xae, 0x56, 0x9e, 0xf3, 0x67, 0xe5, 0x39,
0x3f, 0xd7, 0x5e, 0xe3, 0x6a, 0xed, 0x35, 0x7e, 0xaf, 0xbd, 0xc6, 0x57, 0xd7, 0xde, 0xa3, 0x4a,
0xa7, 0x94, 0x89, 0xe2, 0x95, 0x99, 0x67, 0x92, 0xb4, 0xcd, 0x3b, 0x79, 0xfa, 0x2f, 0x00, 0x00,
0xff, 0xff, 0x2b, 0x2f, 0x77, 0x06, 0x82, 0x03, 0x00, 0x00,
// 374 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x92, 0xdf, 0x8a, 0xd3, 0x40,
0x18, 0xc5, 0x3b, 0xc9, 0x34, 0xce, 0xce, 0xfa, 0x27, 0x0c, 0x75, 0x0d, 0x2e, 0xc4, 0xb0, 0xa8,
0xe4, 0x2a, 0x61, 0xd7, 0x37, 0xa8, 0x20, 0x92, 0x22, 0x48, 0x04, 0x2f, 0x14, 0x0c, 0x93, 0xe4,
0x53, 0xc6, 0x26, 0x99, 0x3a, 0x33, 0x2d, 0xd4, 0xa7, 0xf0, 0x61, 0x7c, 0x08, 0x2f, 0x7b, 0xe9,
0xa5, 0xb4, 0x2f, 0x22, 0x99, 0x04, 0xa5, 0x52, 0xef, 0x72, 0x0e, 0xbf, 0xef, 0xe4, 0xc0, 0x19,
0xfa, 0xb8, 0x92, 0xba, 0x95, 0x3a, 0xd5, 0x46, 0x2a, 0x48, 0x37, 0xd7, 0x25, 0x18, 0x7e, 0x9d,
0x36, 0x42, 0x1b, 0xe8, 0x44, 0xf7, 0x29, 0x59, 0x29, 0x69, 0x24, 0x9b, 0x0d, 0x54, 0x62, 0xa9,
0x64, 0xa4, 0x1e, 0x5e, 0x1a, 0xe8, 0x6a, 0x50, 0xad, 0xe8, 0x4c, 0xca, 0xcb, 0x4a, 0xa4, 0x66,
0xbb, 0x02, 0x3d, 0x9c, 0x5c, 0x7d, 0xa6, 0xe7, 0x6f, 0x7a, 0x7a, 0xf1, 0xf6, 0x35, 0x17, 0x8a,
0x5d, 0xd2, 0x33, 0x7b, 0x5c, 0x2c, 0x61, 0x1b, 0xa0, 0x08, 0xc5, 0x67, 0x39, 0xb1, 0xc6, 0x02,
0xb6, 0xec, 0x82, 0x7a, 0x35, 0x34, 0x60, 0x20, 0x70, 0x22, 0x14, 0x93, 0x7c, 0x54, 0xcc, 0xa7,
0x6e, 0x8f, 0xbb, 0x11, 0x8a, 0x6f, 0xe7, 0xfd, 0x27, 0x9b, 0xd1, 0xe9, 0x86, 0x37, 0x6b, 0x08,
0xb0, 0xf5, 0x06, 0x71, 0xf5, 0xdd, 0xa1, 0x77, 0xe6, 0x8d, 0xac, 0x96, 0xaf, 0xc0, 0xf0, 0x9a,
0x1b, 0xce, 0x5e, 0xd2, 0x7b, 0x0a, 0xf4, 0x4a, 0x76, 0x1a, 0x8a, 0x4a, 0xb6, 0xad, 0x30, 0x81,
0x17, 0xa1, 0xf8, 0xfc, 0xe6, 0x51, 0xf2, 0xb7, 0x74, 0xd2, 0x97, 0x4e, 0xf2, 0x91, 0x7b, 0x6e,
0xb1, 0xfc, 0xae, 0x3a, 0xd2, 0xec, 0x3d, 0xbd, 0x50, 0xf0, 0x65, 0x0d, 0xda, 0x14, 0x1f, 0x45,
0xc7, 0x1b, 0xf1, 0x15, 0x8a, 0xb2, 0xff, 0x57, 0x70, 0xcb, 0x06, 0x3e, 0x39, 0x11, 0x68, 0xf1,
0x17, 0x23, 0x6d, 0x8b, 0xe5, 0x33, 0x75, 0xc2, 0x65, 0x1f, 0xe8, 0x83, 0x3f, 0x35, 0xff, 0x49,
0x27, 0x36, 0xfd, 0xe9, 0x7f, 0xeb, 0x1e, 0xc7, 0xdf, 0x57, 0xa7, 0xec, 0x0c, 0x13, 0xe4, 0x3b,
0x19, 0x26, 0x8e, 0xef, 0x66, 0x98, 0xb8, 0x3e, 0xce, 0x30, 0xc1, 0xfe, 0x34, 0xc3, 0x64, 0xea,
0x7b, 0xf3, 0x9b, 0x1f, 0xfb, 0x10, 0xed, 0xf6, 0x21, 0xfa, 0xb5, 0x0f, 0xd1, 0xb7, 0x43, 0x38,
0xd9, 0x1d, 0xc2, 0xc9, 0xcf, 0x43, 0x38, 0x79, 0x17, 0x0c, 0x7b, 0xeb, 0x7a, 0x99, 0x08, 0x39,
0xbe, 0x0d, 0x3b, 0x6e, 0xe9, 0xd9, 0x75, 0x9f, 0xfd, 0x0e, 0x00, 0x00, 0xff, 0xff, 0x68, 0x61,
0xc9, 0x0c, 0x38, 0x02, 0x00, 0x00,
}
func (m *StoreKVPair) Marshal() (dAtA []byte, err error) {
@@ -352,6 +269,30 @@ func (m *BlockMetadata) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.ResponseFinalizeBlock != nil {
{
size, err := m.ResponseFinalizeBlock.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x42
}
if m.RequestFinalizeBlock != nil {
{
size, err := m.RequestFinalizeBlock.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x3a
}
if m.ResponseCommit != nil {
{
size, err := m.ResponseCommit.MarshalToSizedBuffer(dAtA[:i])
@@ -364,115 +305,6 @@ func (m *BlockMetadata) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i--
dAtA[i] = 0x32
}
if m.ResponseEndBlock != nil {
{
size, err := m.ResponseEndBlock.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x2a
}
if m.RequestEndBlock != nil {
{
size, err := m.RequestEndBlock.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x22
}
if len(m.DeliverTxs) > 0 {
for iNdEx := len(m.DeliverTxs) - 1; iNdEx >= 0; iNdEx-- {
{
size, err := m.DeliverTxs[iNdEx].MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x1a
}
}
if m.ResponseBeginBlock != nil {
{
size, err := m.ResponseBeginBlock.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x12
}
if m.RequestBeginBlock != nil {
{
size, err := m.RequestBeginBlock.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0xa
}
return len(dAtA) - i, nil
}
func (m *BlockMetadata_DeliverTx) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *BlockMetadata_DeliverTx) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *BlockMetadata_DeliverTx) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.Response != nil {
{
size, err := m.Response.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x12
}
if m.Request != nil {
{
size, err := m.Request.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintListening(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0xa
}
return len(dAtA) - i, nil
}
@@ -517,47 +349,16 @@ func (m *BlockMetadata) Size() (n int) {
}
var l int
_ = l
if m.RequestBeginBlock != nil {
l = m.RequestBeginBlock.Size()
n += 1 + l + sovListening(uint64(l))
}
if m.ResponseBeginBlock != nil {
l = m.ResponseBeginBlock.Size()
n += 1 + l + sovListening(uint64(l))
}
if len(m.DeliverTxs) > 0 {
for _, e := range m.DeliverTxs {
l = e.Size()
n += 1 + l + sovListening(uint64(l))
}
}
if m.RequestEndBlock != nil {
l = m.RequestEndBlock.Size()
n += 1 + l + sovListening(uint64(l))
}
if m.ResponseEndBlock != nil {
l = m.ResponseEndBlock.Size()
n += 1 + l + sovListening(uint64(l))
}
if m.ResponseCommit != nil {
l = m.ResponseCommit.Size()
n += 1 + l + sovListening(uint64(l))
}
return n
}
func (m *BlockMetadata_DeliverTx) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if m.Request != nil {
l = m.Request.Size()
if m.RequestFinalizeBlock != nil {
l = m.RequestFinalizeBlock.Size()
n += 1 + l + sovListening(uint64(l))
}
if m.Response != nil {
l = m.Response.Size()
if m.ResponseFinalizeBlock != nil {
l = m.ResponseFinalizeBlock.Size()
n += 1 + l + sovListening(uint64(l))
}
return n
@@ -768,184 +569,6 @@ func (m *BlockMetadata) Unmarshal(dAtA []byte) error {
return fmt.Errorf("proto: BlockMetadata: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field RequestBeginBlock", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowListening
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthListening
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthListening
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.RequestBeginBlock == nil {
m.RequestBeginBlock = &types.RequestBeginBlock{}
}
if err := m.RequestBeginBlock.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field ResponseBeginBlock", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowListening
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthListening
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthListening
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.ResponseBeginBlock == nil {
m.ResponseBeginBlock = &types.ResponseBeginBlock{}
}
if err := m.ResponseBeginBlock.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 3:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field DeliverTxs", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowListening
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthListening
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthListening
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.DeliverTxs = append(m.DeliverTxs, &BlockMetadata_DeliverTx{})
if err := m.DeliverTxs[len(m.DeliverTxs)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 4:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field RequestEndBlock", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowListening
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthListening
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthListening
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.RequestEndBlock == nil {
m.RequestEndBlock = &types.RequestEndBlock{}
}
if err := m.RequestEndBlock.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 5:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field ResponseEndBlock", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowListening
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthListening
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthListening
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.ResponseEndBlock == nil {
m.ResponseEndBlock = &types.ResponseEndBlock{}
}
if err := m.ResponseEndBlock.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 6:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field ResponseCommit", wireType)
@@ -982,59 +605,9 @@ func (m *BlockMetadata) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipListening(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthListening
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *BlockMetadata_DeliverTx) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowListening
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: DeliverTx: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: DeliverTx: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
case 7:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Request", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field RequestFinalizeBlock", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
@@ -1061,16 +634,16 @@ func (m *BlockMetadata_DeliverTx) Unmarshal(dAtA []byte) error {
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.Request == nil {
m.Request = &types.RequestDeliverTx{}
if m.RequestFinalizeBlock == nil {
m.RequestFinalizeBlock = &types.RequestFinalizeBlock{}
}
if err := m.Request.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
if err := m.RequestFinalizeBlock.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 2:
case 8:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Response", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field ResponseFinalizeBlock", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
@@ -1097,10 +670,10 @@ func (m *BlockMetadata_DeliverTx) Unmarshal(dAtA []byte) error {
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.Response == nil {
m.Response = &types.ResponseDeliverTx{}
if m.ResponseFinalizeBlock == nil {
m.ResponseFinalizeBlock = &types.ResponseFinalizeBlock{}
}
if err := m.Response.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
if err := m.ResponseFinalizeBlock.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
store/types/store.go
+1 -1
View File
@@ -40,7 +40,7 @@ type CommitStore interface {
//
// This is an optional, but useful extension to any CommitStore
type Queryable interface {
Query(abci.RequestQuery) abci.ResponseQuery
Query(*abci.RequestQuery) *abci.ResponseQuery
}
//----------------------------------------
store/types/streaming.go
+2 -6
View File
@@ -11,12 +11,8 @@ import (
// The error results are propagated to consensus state machine,
// if you don't want to affect consensus, handle the errors internally and always return `nil` in these APIs.
type ABCIListener interface {
// ListenBeginBlock updates the streaming service with the latest BeginBlock messages
ListenBeginBlock(ctx context.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error
// ListenEndBlock updates the steaming service with the latest EndBlock messages
ListenEndBlock(ctx context.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error
// ListenDeliverTx updates the steaming service with the latest DeliverTx messages
ListenDeliverTx(ctx context.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error
// ListenFinalizeBlock updates the streaming service with the latest FinalizeBlock messages
ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock) error
// ListenCommit updates the steaming service with the latest Commit messages and state changes
ListenCommit(ctx context.Context, res abci.ResponseCommit, changeSet []*StoreKVPair) error
}