add unittests and bug fix

.travis.yml

@@ -8,8 +8,5 @@ before_install:
   - sudo apt-get install redis-server; redis-server &
   - go get github.com/mattn/goveralls
 script:
-  - go test -coverprofile=coverage.txt -covermode=atomic ./...
+  - go test -covermode=atomic ./...
   - $GOPATH/bin/goveralls -service=travis-ci
-
-after_success:
-  - bash <(curl -s https://codecov.io/bash)

cluster/cluster.go

@@ -70,8 +70,8 @@ func MakeCluster() *Cluster {
 	return cluster
 }
 
-// args contains all
-type CmdFunc func(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply
+// CmdFunc represents the handler of a redis command
+type CmdFunc func(cluster *Cluster, c redis.Connection, cmdAndArgs [][]byte) redis.Reply
 
 func (cluster *Cluster) Close() {
 	cluster.db.Close()

cluster/pubsub.go

@@ -19,20 +19,11 @@ var (
 // broadcast msg to all peers in cluster when receive publish command from client
 func Publish(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
 	var count int64 = 0
-	for _, peer := range cluster.nodes {
-		var re redis.Reply
-		if peer == cluster.self {
-			args0 := make([][]byte, len(args))
-			copy(args0, args)
-			args0[0] = publishCmd
-			re = cluster.db.Exec(c, args0) // let local db.hub handle publish
-		} else {
-			args[0] = publishRelayCmd
-			re = cluster.Relay(peer, c, args)
-		}
-		if errReply, ok := re.(reply.ErrorReply); ok {
+	results := cluster.Broadcast(c, args)
+	for _, val := range results {
+		if errReply, ok := val.(reply.ErrorReply); ok {
 			logger.Error("publish occurs error: " + errReply.Error())
-		} else if intReply, ok := re.(*reply.IntReply); ok {
+		} else if intReply, ok := val.(*reply.IntReply); ok {
 			count += intReply.Code
 		}
 	}

cluster/pubsub_test.go

@@ -0,0 +1,48 @@
+package cluster
+
+import (
+	"github.com/hdt3213/godis/lib/utils"
+	"github.com/hdt3213/godis/redis/connection"
+	"github.com/hdt3213/godis/redis/parser"
+	"github.com/hdt3213/godis/redis/reply/asserts"
+	"testing"
+)
+
+func TestPublish(t *testing.T) {
+	channel := utils.RandString(5)
+	msg := utils.RandString(5)
+	conn := &connection.FakeConn{}
+	Subscribe(testCluster, conn, utils.ToBytesList("SUBSCRIBE", channel))
+	conn.Clean() // clean subscribe success
+	Publish(testCluster, conn, utils.ToBytesList("PUBLISH", channel, msg))
+	data := conn.Bytes()
+	ret, err := parser.ParseOne(data)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	asserts.AssertMultiBulkReply(t, ret, []string{
+		"message",
+		channel,
+		msg,
+	})
+
+	// unsubscribe
+	UnSubscribe(testCluster, conn, utils.ToBytesList("UNSUBSCRIBE", channel))
+	conn.Clean()
+	Publish(testCluster, conn, utils.ToBytesList("PUBLISH", channel, msg))
+	data = conn.Bytes()
+	if len(data) > 0 {
+		t.Error("expect no msg")
+	}
+
+	// unsubscribe all
+	Subscribe(testCluster, conn, utils.ToBytesList("SUBSCRIBE", channel))
+	UnSubscribe(testCluster, conn, utils.ToBytesList("UNSUBSCRIBE"))
+	conn.Clean()
+	Publish(testCluster, conn, utils.ToBytesList("PUBLISH", channel, msg))
+	data = conn.Bytes()
+	if len(data) > 0 {
+		t.Error("expect no msg")
+	}
+}

db/aof.go

@@ -82,7 +82,7 @@ func (db *DB) loadAof(maxBytes int) {
 	defer file.Close()
 
 	reader := utils.NewLimitedReader(file, maxBytes)
-	ch := parser.Parse(reader)
+	ch := parser.ParseStream(reader)
 	for p := range ch {
 		if p.Err != nil {
 			if p.Err == io.EOF {

db/aof_test.go

@@ -3,6 +3,8 @@ package db
 import (
 	"github.com/hdt3213/godis/config"
 	"github.com/hdt3213/godis/datastruct/utils"
+	utils2 "github.com/hdt3213/godis/lib/utils"
+	"github.com/hdt3213/godis/redis/reply"
 	"io/ioutil"
 	"os"
 	"path"
@@ -32,31 +34,31 @@ func TestAof(t *testing.T) {
 	for i := 0; i < size; i++ {
 		key := strconv.Itoa(cursor)
 		cursor++
-		Set(aofWriteDB, toArgs(key, RandString(8), "EX", "10000"))
+		Set(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8), "EX", "10000"))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := strconv.Itoa(cursor)
 		cursor++
-		RPush(aofWriteDB, toArgs(key, RandString(8)))
+		RPush(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := strconv.Itoa(cursor)
 		cursor++
-		HSet(aofWriteDB, toArgs(key, RandString(8), RandString(8)))
+		HSet(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8), utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := strconv.Itoa(cursor)
 		cursor++
-		SAdd(aofWriteDB, toArgs(key, RandString(8)))
+		SAdd(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := strconv.Itoa(cursor)
 		cursor++
-		ZAdd(aofWriteDB, toArgs(key, "10", RandString(8)))
+		ZAdd(aofWriteDB, utils2.ToBytesList(key, "10", utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	aofWriteDB.Close()    // wait for aof finished
@@ -98,41 +100,49 @@ func TestRewriteAOF(t *testing.T) {
 	aofWriteDB := MakeDB()
 	size := 1
 	keys := make([]string, 0)
+	ttlKeys := make([]string, 0)
 	cursor := 0
 	for i := 0; i < size; i++ {
 		key := "str" + strconv.Itoa(cursor)
 		cursor++
-		Set(aofWriteDB, toArgs(key, RandString(8)))
-		Set(aofWriteDB, toArgs(key, RandString(8)))
+		Set(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8)))
+		Set(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8)))
 		keys = append(keys, key)
 	}
+	// test ttl
+	for i := 0; i < size; i++ {
+		key := "str" + strconv.Itoa(cursor)
+		cursor++
+		Set(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8), "EX", "1000"))
+		ttlKeys = append(ttlKeys, key)
+	}
 	for i := 0; i < size; i++ {
 		key := "list" + strconv.Itoa(cursor)
 		cursor++
-		RPush(aofWriteDB, toArgs(key, RandString(8)))
-		RPush(aofWriteDB, toArgs(key, RandString(8)))
+		RPush(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8)))
+		RPush(aofWriteDB, utils2.ToBytesList(key, utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := "hash" + strconv.Itoa(cursor)
 		cursor++
-		field := RandString(8)
-		HSet(aofWriteDB, toArgs(key, field, RandString(8)))
-		HSet(aofWriteDB, toArgs(key, field, RandString(8)))
+		field := utils2.RandString(8)
+		HSet(aofWriteDB, utils2.ToBytesList(key, field, utils2.RandString(8)))
+		HSet(aofWriteDB, utils2.ToBytesList(key, field, utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := "set" + strconv.Itoa(cursor)
 		cursor++
-		member := RandString(8)
-		SAdd(aofWriteDB, toArgs(key, member))
-		SAdd(aofWriteDB, toArgs(key, member))
+		member := utils2.RandString(8)
+		SAdd(aofWriteDB, utils2.ToBytesList(key, member))
+		SAdd(aofWriteDB, utils2.ToBytesList(key, member))
 		keys = append(keys, key)
 	}
 	for i := 0; i < size; i++ {
 		key := "zset" + strconv.Itoa(cursor)
 		cursor++
-		ZAdd(aofWriteDB, toArgs(key, "10", RandString(8)))
+		ZAdd(aofWriteDB, utils2.ToBytesList(key, "10", utils2.RandString(8)))
 		keys = append(keys, key)
 	}
 	time.Sleep(time.Second) // wait for async goroutine finish its job
@@ -156,5 +166,16 @@ func TestRewriteAOF(t *testing.T) {
 			t.Errorf("wrong value of key: %s", key)
 		}
 	}
+	for _, key := range ttlKeys {
+		ret := TTL(aofReadDB, utils2.ToBytesList(key))
+		intResult, ok := ret.(*reply.IntReply)
+		if !ok {
+			t.Errorf("expected int reply, actually %s", ret.ToBytes())
+			return
+		}
+		if intResult.Code <= 0 {
+			t.Errorf("expect a positive integer, actual: %d", intResult.Code)
+		}
+	}
 	aofReadDB.Close()
 }

db/geo_test.go

@@ -2,75 +2,76 @@ package db
 
 import (
 	"fmt"
+	"github.com/hdt3213/godis/lib/utils"
 	"github.com/hdt3213/godis/redis/reply"
 	"github.com/hdt3213/godis/redis/reply/asserts"
 	"strconv"
 	"testing"
 )
 
 func TestGeoHash(t *testing.T) {
-	FlushDB(testDB, toArgs())
-	key := RandString(10)
-	pos := RandString(10)
-	result := GeoAdd(testDB, toArgs(key, "13.361389", "38.115556", pos))
+	FlushDB(testDB, utils.ToBytesList())
+	key := utils.RandString(10)
+	pos := utils.RandString(10)
+	result := GeoAdd(testDB, utils.ToBytesList(key, "13.361389", "38.115556", pos))
 	asserts.AssertIntReply(t, result, 1)
-	result = GeoHash(testDB, toArgs(key, pos))
+	result = GeoHash(testDB, utils.ToBytesList(key, pos))
 	asserts.AssertMultiBulkReply(t, result, []string{"sqc8b49rnys00"})
 }
 
 func TestGeoRadius(t *testing.T) {
-	FlushDB(testDB, toArgs())
-	key := RandString(10)
-	pos1 := RandString(10)
-	pos2 := RandString(10)
-	GeoAdd(testDB, toArgs(key,
+	FlushDB(testDB, utils.ToBytesList())
+	key := utils.RandString(10)
+	pos1 := utils.RandString(10)
+	pos2 := utils.RandString(10)
+	GeoAdd(testDB, utils.ToBytesList(key,
 		"13.361389", "38.115556", pos1,
 		"15.087269", "37.502669", pos2,
 	))
-	result := GeoRadius(testDB, toArgs(key, "15", "37", "200", "km"))
+	result := GeoRadius(testDB, utils.ToBytesList(key, "15", "37", "200", "km"))
 	asserts.AssertMultiBulkReplySize(t, result, 2)
 }
 
 func TestGeoRadiusByMember(t *testing.T) {
-	FlushDB(testDB, toArgs())
-	key := RandString(10)
-	pos1 := RandString(10)
-	pos2 := RandString(10)
-	pivot := RandString(10)
-	GeoAdd(testDB, toArgs(key,
+	FlushDB(testDB, utils.ToBytesList())
+	key := utils.RandString(10)
+	pos1 := utils.RandString(10)
+	pos2 := utils.RandString(10)
+	pivot := utils.RandString(10)
+	GeoAdd(testDB, utils.ToBytesList(key,
 		"13.361389", "38.115556", pos1,
 		"17.087269", "38.502669", pos2,
 		"13.583333", "37.316667", pivot,
 	))
-	result := GeoRadiusByMember(testDB, toArgs(key, pivot, "100", "km"))
+	result := GeoRadiusByMember(testDB, utils.ToBytesList(key, pivot, "100", "km"))
 	asserts.AssertMultiBulkReplySize(t, result, 2)
 }
 
 func TestGeoPos(t *testing.T) {
-	FlushDB(testDB, toArgs())
-	key := RandString(10)
-	pos1 := RandString(10)
-	pos2 := RandString(10)
-	GeoAdd(testDB, toArgs(key,
+	FlushDB(testDB, utils.ToBytesList())
+	key := utils.RandString(10)
+	pos1 := utils.RandString(10)
+	pos2 := utils.RandString(10)
+	GeoAdd(testDB, utils.ToBytesList(key,
 		"13.361389", "38.115556", pos1,
 	))
-	result := GeoPos(testDB, toArgs(key, pos1, pos2))
+	result := GeoPos(testDB, utils.ToBytesList(key, pos1, pos2))
 	expected := "*2\r\n*2\r\n$18\r\n13.361386698670685\r\n$17\r\n38.11555536696687\r\n*0\r\n"
 	if string(result.ToBytes()) != expected {
 		t.Error("test failed")
 	}
 }
 
 func TestGeoDist(t *testing.T) {
-	FlushDB(testDB, toArgs())
-	key := RandString(10)
-	pos1 := RandString(10)
-	pos2 := RandString(10)
-	GeoAdd(testDB, toArgs(key,
+	FlushDB(testDB, utils.ToBytesList())
+	key := utils.RandString(10)
+	pos1 := utils.RandString(10)
+	pos2 := utils.RandString(10)
+	GeoAdd(testDB, utils.ToBytesList(key,
 		"13.361389", "38.115556", pos1,
 		"15.087269", "37.502669", pos2,
 	))
-	result := GeoDist(testDB, toArgs(key, pos1, pos2, "km"))
+	result := GeoDist(testDB, utils.ToBytesList(key, pos1, pos2, "km"))
 	bulkReply, ok := result.(*reply.BulkReply)
 	if !ok {
 		t.Error(fmt.Sprintf("expected bulk reply, actually %s", result.ToBytes()))