Fixing RedisRTC implementation and test

go/redisutil/rtc_test.go

Index: go/redisutil/rtc_test.go
diff --git a/go/redisutil/rtc_test.go b/go/redisutil/rtc_test.go
index c3e0494cb7f64d025bdf6c3fcacb3a1885923b4a..7b5584b839df69464441ebc0477f082f2bd9010c 100644
--- a/go/redisutil/rtc_test.go
+++ b/go/redisutil/rtc_test.go
@@ -1,49 +1,61 @@
 package redisutil
 
 import (
+	"crypto/rand"
 	"fmt"
 	"runtime"
-	"strconv"
 	"sync"
 	"testing"
 
-	"go.skia.org/infra/go/testutils"
-
 	assert "github.com/stretchr/testify/require"
+
+	"go.skia.org/infra/go/testutils"
 )
 
 const (
 	Q_NAME            = "mytype"
 	Q_NAME_PRIMITIVES = "mytestq"
-	N_TASKS           = 10000
+	N_TASKS           = 1

[dogben, 2016/08/08 21:18:24]

Did you intend to test only one task? Seems like the higher number of tasks was
the only reason we caught the issue with not closing.

[stephana, 2016/08/08 21:46:55]

Good catch. That was for debugging. Changed to 1000. 

+	PACKAGE_SIZE      = 1024 * 512
 )
 
+// BytesCodec for testing.
+type BytesCodec struct{}
+
+func (b BytesCodec) Encode(data interface{}) ([]byte, error) {
+	// Make a copy to simulate the generic case.
+	return append([]byte(nil), data.([]byte)...), nil
+}
+
+func (b BytesCodec) Decode(byteData []byte) (interface{}, error) {
+	// Make a copy to simulate the generic case.
+	return append([]byte(nil), byteData...), nil
+}
+
 // TODO (stephana): Re-enable this test once we have a way to cleanly shutdown

[dogben, 2016/08/08 21:18:24]

delete?

[stephana, 2016/08/08 21:46:55]

Done.

 // an instance of RedisRTC.
-func xTestReadThroughCache(t *testing.T) {
+func TestReadThroughCache(t *testing.T) {
 	testutils.SkipIfShort(t)
 
-	runtime.GOMAXPROCS(runtime.NumCPU() - 1)
-
 	rp := NewRedisPool(REDIS_SERVER_ADDRESS, REDIS_DB_RTCACHE)
+	defer testutils.CloseInTest(t, rp)
+
 	assert.NoError(t, rp.FlushDB())
+	randBytes := make([]byte, PACKAGE_SIZE)
+	_, err := rand.Read(randBytes)
 

[dogben, 2016/08/08 21:18:24]

Do you want assert.NoError here, and not return err below?

[stephana, 2016/08/08 21:46:55]

Done.

 	worker := func(priority int64, id string) (interface{}, error) {
-		// Run a few calculations in a loop.
-		result := 0
-		for i := 0; i < 10; i++ {
-			result += i
-		}
-
-		// Do the work
-		return id + "-" + strconv.Itoa(result), nil
+		// Create a unique version of the random array.
+		return []byte(id + string(randBytes)), err
 	}
 
 	// create a worker queue for a given type
-	codec := StringCodec{}
-	qRet, err := NewReadThroughCache(rp, Q_NAME, nil, codec, runtime.NumCPU()-2)
+	// codec := StringCodec{}

[dogben, 2016/08/08 21:18:24]

nit: delete

[stephana, 2016/08/08 21:46:55]

Done.

+	codec := BytesCodec{}
+	qRet, err := NewReadThroughCache(rp, Q_NAME, worker, codec, runtime.NumCPU()-2)
 	assert.NoError(t, err)
 	q := qRet.(*RedisRTC)
+	defer q.shutdown()
 
 	// make sure all results arrive.
 	var allDone sync.WaitGroup
@@ -53,7 +65,7 @@ func xTestReadThroughCache(t *testing.T) {
 	for i := 0; i < N_TASKS; i++ {
 		allDone.Add(1)
 		go func(idx, priority int) {
-			id := "id-" + strconv.Itoa(idx)
+			id := "id-" + fmt.Sprintf("%04d", idx)
 			result, err := q.Get(int64(priority), false, id)
 			if err != nil {
 				errCh <- err
@@ -64,9 +76,6 @@ func xTestReadThroughCache(t *testing.T) {
 			allDone.Done()
 		}(i, i)
 	}
-
-	q.worker = worker
-	assert.NoError(t, q.startWorkers(runtime.NumCPU()-2))
 	allDone.Wait()
 
 	close(errCh)
@@ -76,15 +85,17 @@ func xTestReadThroughCache(t *testing.T) {
 		for err := range errCh {
 			fmt.Printf("Error: %s", err)
 		}
-		assert.True(t, false)
+		assert.Fail(t, "Received above error messages.")
 	}
 
 	assert.Equal(t, 0, len(errCh))
 	found := make(map[string]bool, N_TASKS)
 	for ret := range retCh {
-		assert.IsType(t, "", ret)
-		found[ret.(string)] = true
+		assert.IsType(t, []byte(""), ret)
+		assert.Equal(t, 1024*512+7, len(ret.([]byte)))

[dogben, 2016/08/08 21:18:24]

nit: maybe add a comment to explain this value

[stephana, 2016/08/08 21:46:55]

Done.

+		found[string(ret.([]byte))] = true
 	}
 
+	// Make sure all strings are unique.
 	assert.Equal(t, N_TASKS, len(found))
 }