vendor/github.com/tendermint/abci/example/example_test.go

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