Added counter test written in Golang

example/dummy/dummy.go

@@ -0,0 +1,54 @@
+package dummy
+
+import (
+	"strings"
+
+	. "github.com/tendermint/go-common"
+	"github.com/tendermint/go-merkle"
+	"github.com/tendermint/tmsp/types"
+)
+
+type DummyApplication struct {
+	state merkle.Tree
+}
+
+func NewDummyApplication() *DummyApplication {
+	state := merkle.NewIAVLTree(
+		0,
+		nil,
+	)
+	return &DummyApplication{state: state}
+}
+
+func (app *DummyApplication) Info() string {
+	return Fmt("size:%v", app.state.Size())
+}
+
+func (app *DummyApplication) SetOption(key string, value string) (log string) {
+	return ""
+}
+
+func (app *DummyApplication) AppendTx(tx []byte) (code types.CodeType, result []byte, log string) {
+	parts := strings.Split(string(tx), "=")
+	if len(parts) == 2 {
+		app.state.Set([]byte(parts[0]), []byte(parts[1]))
+	} else {
+		app.state.Set(tx, tx)
+	}
+	return types.CodeType_OK, nil, ""
+}
+
+func (app *DummyApplication) CheckTx(tx []byte) (code types.CodeType, result []byte, log string) {
+	return types.CodeType_OK, nil, ""
+}
+
+func (app *DummyApplication) Commit() (hash []byte, log string) {
+	hash = app.state.Hash()
+	return hash, ""
+}
+
+func (app *DummyApplication) Query(query []byte) (code types.CodeType, result []byte, log string) {
+	index, value, exists := app.state.Get(query)
+	resStr := Fmt("Index=%v value=%v exists=%v", index, string(value), exists)
+	return types.CodeType_OK, []byte(resStr), ""
+}

example/dummy/dummy_test.go

@@ -0,0 +1,91 @@
+package dummy
+
+import (
+	"testing"
+	"time"
+
+	. "github.com/tendermint/go-common"
+	"github.com/tendermint/tmsp/server"
+	"github.com/tendermint/tmsp/types"
+)
+
+func TestStream(t *testing.T) {
+
+	numAppendTxs := 200000
+
+	// Start the listener
+	_, err := server.StartListener("tcp://127.0.0.1:46658", NewDummyApplication())
+	if err != nil {
+		Exit(err.Error())
+	}
+
+	// Connect to the socket
+	conn, err := Connect("tcp://127.0.0.1:46658")
+	if err != nil {
+		Exit(err.Error())
+	}
+
+	// Read response data
+	done := make(chan struct{})
+	go func() {
+		counter := 0
+		for {
+
+			var res = &types.Response{}
+			err := types.ReadMessage(conn, res)
+			if err != nil {
+				Exit(err.Error())
+			}
+
+			// Process response
+			switch res.Type {
+			case types.MessageType_AppendTx:
+				counter += 1
+				if res.Code != types.CodeType_OK {
+					t.Error("AppendTx failed with ret_code", res.Code)
+				}
+				if counter > numAppendTxs {
+					t.Fatal("Too many AppendTx responses")
+				}
+				t.Log("response", counter)
+				if counter == numAppendTxs {
+					go func() {
+						time.Sleep(time.Second * 2) // Wait for a bit to allow counter overflow
+						close(done)
+					}()
+				}
+			case types.MessageType_Flush:
+				// ignore
+			default:
+				t.Error("Unexpected response type", res.Type)
+			}
+		}
+	}()
+
+	// Write requests
+	for counter := 0; counter < numAppendTxs; counter++ {
+		// Send request
+		var req = types.RequestAppendTx([]byte("test"))
+		err := types.WriteMessage(req, conn)
+		if err != nil {
+			t.Fatal(err.Error())
+		}
+
+		// Sometimes send flush messages
+		if counter%123 == 0 {
+			t.Log("flush")
+			err := types.WriteMessage(types.RequestFlush(), conn)
+			if err != nil {
+				t.Fatal(err.Error())
+			}
+		}
+	}
+
+	// Send final flush message
+	err = types.WriteMessage(types.RequestFlush(), conn)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	<-done
+}