mirror of
https://github.com/fluencelabs/tendermint
synced 2025-04-25 06:42:16 +00:00
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
157 lines
4.0 KiB
Go
157 lines
4.0 KiB
Go
package example
|
|
|
|
import (
|
|
"fmt"
|
|
"net"
|
|
"reflect"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"google.golang.org/grpc"
|
|
|
|
"golang.org/x/net/context"
|
|
|
|
cmn "github.com/tendermint/tendermint/libs/common"
|
|
"github.com/tendermint/tendermint/libs/log"
|
|
|
|
abcicli "github.com/tendermint/tendermint/abci/client"
|
|
"github.com/tendermint/tendermint/abci/example/code"
|
|
"github.com/tendermint/tendermint/abci/example/kvstore"
|
|
abciserver "github.com/tendermint/tendermint/abci/server"
|
|
"github.com/tendermint/tendermint/abci/types"
|
|
)
|
|
|
|
func TestKVStore(t *testing.T) {
|
|
fmt.Println("### Testing KVStore")
|
|
testStream(t, kvstore.NewKVStoreApplication())
|
|
}
|
|
|
|
func TestBaseApp(t *testing.T) {
|
|
fmt.Println("### Testing BaseApp")
|
|
testStream(t, types.NewBaseApplication())
|
|
}
|
|
|
|
func TestGRPC(t *testing.T) {
|
|
fmt.Println("### Testing GRPC")
|
|
testGRPCSync(t, types.NewGRPCApplication(types.NewBaseApplication()))
|
|
}
|
|
|
|
func testStream(t *testing.T, app types.Application) {
|
|
numDeliverTxs := 20000
|
|
|
|
// Start the listener
|
|
server := abciserver.NewSocketServer("unix://test.sock", app)
|
|
server.SetLogger(log.TestingLogger().With("module", "abci-server"))
|
|
if err := server.Start(); err != nil {
|
|
require.NoError(t, err, "Error starting socket server")
|
|
}
|
|
defer server.Stop()
|
|
|
|
// Connect to the socket
|
|
client := abcicli.NewSocketClient("unix://test.sock", false)
|
|
client.SetLogger(log.TestingLogger().With("module", "abci-client"))
|
|
if err := client.Start(); err != nil {
|
|
t.Fatalf("Error starting socket client: %v", err.Error())
|
|
}
|
|
defer client.Stop()
|
|
|
|
done := make(chan struct{})
|
|
counter := 0
|
|
client.SetResponseCallback(func(req *types.Request, res *types.Response) {
|
|
// Process response
|
|
switch r := res.Value.(type) {
|
|
case *types.Response_DeliverTx:
|
|
counter++
|
|
if r.DeliverTx.Code != code.CodeTypeOK {
|
|
t.Error("DeliverTx failed with ret_code", r.DeliverTx.Code)
|
|
}
|
|
if counter > numDeliverTxs {
|
|
t.Fatalf("Too many DeliverTx responses. Got %d, expected %d", counter, numDeliverTxs)
|
|
}
|
|
if counter == numDeliverTxs {
|
|
go func() {
|
|
time.Sleep(time.Second * 1) // Wait for a bit to allow counter overflow
|
|
close(done)
|
|
}()
|
|
return
|
|
}
|
|
case *types.Response_Flush:
|
|
// ignore
|
|
default:
|
|
t.Error("Unexpected response type", reflect.TypeOf(res.Value))
|
|
}
|
|
})
|
|
|
|
// Write requests
|
|
for counter := 0; counter < numDeliverTxs; counter++ {
|
|
// Send request
|
|
reqRes := client.DeliverTxAsync(types.RequestDeliverTx{Tx: []byte("test")})
|
|
_ = reqRes
|
|
// check err ?
|
|
|
|
// Sometimes send flush messages
|
|
if counter%123 == 0 {
|
|
client.FlushAsync()
|
|
// check err ?
|
|
}
|
|
}
|
|
|
|
// Send final flush message
|
|
client.FlushAsync()
|
|
|
|
<-done
|
|
}
|
|
|
|
//-------------------------
|
|
// test grpc
|
|
|
|
func dialerFunc(addr string, timeout time.Duration) (net.Conn, error) {
|
|
return cmn.Connect(addr)
|
|
}
|
|
|
|
func testGRPCSync(t *testing.T, app *types.GRPCApplication) {
|
|
numDeliverTxs := 2000
|
|
|
|
// Start the listener
|
|
server := abciserver.NewGRPCServer("unix://test.sock", app)
|
|
server.SetLogger(log.TestingLogger().With("module", "abci-server"))
|
|
if err := server.Start(); err != nil {
|
|
t.Fatalf("Error starting GRPC server: %v", err.Error())
|
|
}
|
|
defer server.Stop()
|
|
|
|
// Connect to the socket
|
|
conn, err := grpc.Dial("unix://test.sock", grpc.WithInsecure(), grpc.WithDialer(dialerFunc))
|
|
if err != nil {
|
|
t.Fatalf("Error dialing GRPC server: %v", err.Error())
|
|
}
|
|
defer conn.Close()
|
|
|
|
client := types.NewABCIApplicationClient(conn)
|
|
|
|
// Write requests
|
|
for counter := 0; counter < numDeliverTxs; counter++ {
|
|
// Send request
|
|
response, err := client.DeliverTx(context.Background(), &types.RequestDeliverTx{Tx: []byte("test")})
|
|
if err != nil {
|
|
t.Fatalf("Error in GRPC DeliverTx: %v", err.Error())
|
|
}
|
|
counter++
|
|
if response.Code != code.CodeTypeOK {
|
|
t.Error("DeliverTx failed with ret_code", response.Code)
|
|
}
|
|
if counter > numDeliverTxs {
|
|
t.Fatal("Too many DeliverTx responses")
|
|
}
|
|
t.Log("response", counter)
|
|
if counter == numDeliverTxs {
|
|
go func() {
|
|
time.Sleep(time.Second * 1) // Wait for a bit to allow counter overflow
|
|
}()
|
|
}
|
|
|
|
}
|
|
}
|