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
4.0 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package example
  2. import (
  3. "fmt"
  4. "net"
  5. "reflect"
  6. "testing"
  7. "time"
  8. "github.com/stretchr/testify/require"
  9. "google.golang.org/grpc"
  10. "golang.org/x/net/context"
  11. cmn "github.com/tendermint/tendermint/libs/common"
  12. "github.com/tendermint/tendermint/libs/log"
  13. abcicli "github.com/tendermint/tendermint/abci/client"
  14. "github.com/tendermint/tendermint/abci/example/code"
  15. "github.com/tendermint/tendermint/abci/example/kvstore"
  16. abciserver "github.com/tendermint/tendermint/abci/server"
  17. "github.com/tendermint/tendermint/abci/types"
  18. )
  19. func TestKVStore(t *testing.T) {
  20. fmt.Println("### Testing KVStore")
  21. testStream(t, kvstore.NewKVStoreApplication())
  22. }
  23. func TestBaseApp(t *testing.T) {
  24. fmt.Println("### Testing BaseApp")
  25. testStream(t, types.NewBaseApplication())
  26. }
  27. func TestGRPC(t *testing.T) {
  28. fmt.Println("### Testing GRPC")
  29. testGRPCSync(t, types.NewGRPCApplication(types.NewBaseApplication()))
  30. }
  31. func testStream(t *testing.T, app types.Application) {
  32. numDeliverTxs := 20000
  33. // Start the listener
  34. server := abciserver.NewSocketServer("unix://test.sock", app)
  35. server.SetLogger(log.TestingLogger().With("module", "abci-server"))
  36. if err := server.Start(); err != nil {
  37. require.NoError(t, err, "Error starting socket server")
  38. }
  39. defer server.Stop()
  40. // Connect to the socket
  41. client := abcicli.NewSocketClient("unix://test.sock", false)
  42. client.SetLogger(log.TestingLogger().With("module", "abci-client"))
  43. if err := client.Start(); err != nil {
  44. t.Fatalf("Error starting socket client: %v", err.Error())
  45. }
  46. defer client.Stop()
  47. done := make(chan struct{})
  48. counter := 0
  49. client.SetResponseCallback(func(req *types.Request, res *types.Response) {
  50. // Process response
  51. switch r := res.Value.(type) {
  52. case *types.Response_DeliverTx:
  53. counter++
  54. if r.DeliverTx.Code != code.CodeTypeOK {
  55. t.Error("DeliverTx failed with ret_code", r.DeliverTx.Code)
  56. }
  57. if counter > numDeliverTxs {
  58. t.Fatalf("Too many DeliverTx responses. Got %d, expected %d", counter, numDeliverTxs)
  59. }
  60. if counter == numDeliverTxs {
  61. go func() {
  62. time.Sleep(time.Second * 1) // Wait for a bit to allow counter overflow
  63. close(done)
  64. }()
  65. return
  66. }
  67. case *types.Response_Flush:
  68. // ignore
  69. default:
  70. t.Error("Unexpected response type", reflect.TypeOf(res.Value))
  71. }
  72. })
  73. // Write requests
  74. for counter := 0; counter < numDeliverTxs; counter++ {
  75. // Send request
  76. reqRes := client.DeliverTxAsync(types.RequestDeliverTx{Tx: []byte("test")})
  77. _ = reqRes
  78. // check err ?
  79. // Sometimes send flush messages
  80. if counter%123 == 0 {
  81. client.FlushAsync()
  82. // check err ?
  83. }
  84. }
  85. // Send final flush message
  86. client.FlushAsync()
  87. <-done
  88. }
  89. //-------------------------
  90. // test grpc
  91. func dialerFunc(addr string, timeout time.Duration) (net.Conn, error) {
  92. return cmn.Connect(addr)
  93. }
  94. func testGRPCSync(t *testing.T, app *types.GRPCApplication) {
  95. numDeliverTxs := 2000
  96. // Start the listener
  97. server := abciserver.NewGRPCServer("unix://test.sock", app)
  98. server.SetLogger(log.TestingLogger().With("module", "abci-server"))
  99. if err := server.Start(); err != nil {
  100. t.Fatalf("Error starting GRPC server: %v", err.Error())
  101. }
  102. defer server.Stop()
  103. // Connect to the socket
  104. conn, err := grpc.Dial("unix://test.sock", grpc.WithInsecure(), grpc.WithDialer(dialerFunc))
  105. if err != nil {
  106. t.Fatalf("Error dialing GRPC server: %v", err.Error())
  107. }
  108. defer conn.Close()
  109. client := types.NewABCIApplicationClient(conn)
  110. // Write requests
  111. for counter := 0; counter < numDeliverTxs; counter++ {
  112. // Send request
  113. response, err := client.DeliverTx(context.Background(), &types.RequestDeliverTx{Tx: []byte("test")})
  114. if err != nil {
  115. t.Fatalf("Error in GRPC DeliverTx: %v", err.Error())
  116. }
  117. counter++
  118. if response.Code != code.CodeTypeOK {
  119. t.Error("DeliverTx failed with ret_code", response.Code)
  120. }
  121. if counter > numDeliverTxs {
  122. t.Fatal("Too many DeliverTx responses")
  123. }
  124. t.Log("response", counter)
  125. if counter == numDeliverTxs {
  126. go func() {
  127. time.Sleep(time.Second * 1) // Wait for a bit to allow counter overflow
  128. }()
  129. }
  130. }
  131. }