test: grpc_client_test and fix grpc_client StopForError() hang bug

abci/client/grpc_client_test.go

@@ -0,0 +1,102 @@
+package abcicli_test
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	abcicli "github.com/line/ostracon/abci/client"
+	"github.com/line/ostracon/abci/server"
+	"github.com/line/ostracon/abci/types"
+	ostrand "github.com/line/ostracon/libs/rand"
+	"github.com/line/ostracon/libs/service"
+)
+
+type errorStopper interface {
+	StopForError(error)
+}
+
+func TestSocketClientStopForErrorDeadlock(t *testing.T) {
+	c := abcicli.NewGRPCClient(":80", false).(errorStopper)
+	err := errors.New("foo-ostracon")
+
+	// See Issue https://github.com/tendermint/abci/issues/114
+	doneChan := make(chan bool)
+	go func() {
+		defer close(doneChan)
+		c.StopForError(err)
+		c.StopForError(err)
+	}()
+
+	select {
+	case <-doneChan:
+	case <-time.After(time.Second * 4):
+		t.Fatalf("Test took too long, potential deadlock still exists")
+	}
+}
+
+func TestProperSyncCalls(t *testing.T) {
+	app := slowApp{}
+
+	s, c := setupClientServer(t, app)
+	t.Cleanup(func() {
+		if err := s.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+	t.Cleanup(func() {
+		if err := c.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+
+	resp := make(chan error, 1)
+	go func() {
+		// This is BeginBlockSync unrolled....
+		reqres := c.BeginBlockAsync(types.RequestBeginBlock{})
+		err := c.FlushSync()
+		require.NoError(t, err)
+		res := reqres.Response.GetBeginBlock()
+		require.NotNil(t, res)
+		resp <- c.Error()
+	}()
+
+	select {
+	case <-time.After(time.Second):
+		require.Fail(t, "No response arrived")
+	case err, ok := <-resp:
+		require.True(t, ok, "Must not close channel")
+		assert.NoError(t, err, "This should return success")
+	}
+}
+
+func setupClientServer(t *testing.T, app types.Application) (
+	service.Service, abcicli.Client) {
+	// some port between 20k and 30k
+	port := 20000 + ostrand.Int32()%10000
+	addr := fmt.Sprintf("localhost:%d", port)
+
+	s, err := server.NewServer(addr, "grpc", app)
+	require.NoError(t, err)
+	err = s.Start()
+	require.NoError(t, err)
+
+	c := abcicli.NewGRPCClient(addr, true)
+	err = c.Start()
+	require.NoError(t, err)
+
+	return s, c
+}
+
+type slowApp struct {
+	types.BaseApplication
+}
+
+func (slowApp) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginBlock {
+	time.Sleep(200 * time.Millisecond)
+	return types.ResponseBeginBlock{}
+}