use lever instead of flagconfig, use radix.v2 instead of radix, fix Lua call for cluster

Author: Brian Picciano

.go.yaml

@@ -1,7 +1,7 @@
 ---
 path: github.com/mc0/okq
 deps:
-    - loc: github.com/fzzy/radix/redis
-    - loc: github.com/mediocregopher/flagconfig
+    - loc: github.com/mediocregopher/lever
     - loc: github.com/mediocregopher/pubsubch
+    - loc: github.com/mediocregopher/radix.v2
     - loc: github.com/stretchr/testify

clients/consumers/consumers_test.go

@@ -4,7 +4,7 @@ import (
 	. "testing"
 	"time"
 
-	"github.com/fzzy/radix/redis"
+	"github.com/mediocregopher/radix.v2/redis"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -27,7 +27,7 @@ func TestUpdateQueues(t *T) {
 	for i := range queues {
 		key := db.ConsumersKey(queues[i])
 		res := db.Cmd("ZRANK", key, client.ID)
-		assert.Equal(t, redis.IntegerReply, res.Type, "res: %s", res)
+		assert.Equal(t, true, res.IsType(redis.Int), "res: %s", res)
 	}
 
 	err = UpdateQueues(client, queues[1:])
@@ -36,13 +36,13 @@ func TestUpdateQueues(t *T) {
 	// Make sure the first queue had this clientId removed from it
 	key := db.ConsumersKey(queues[0])
 	res := db.Cmd("ZRANK", key, client.ID)
-	assert.Equal(t, redis.NilReply, res.Type, "res: %s", res)
+	assert.Equal(t, true, res.IsType(redis.Nil), "res: %s", res)
 
 	// Make sure the rest of the queues still have it
 	for i := range queues[1:] {
 		key := db.ConsumersKey(queues[1:][i])
 		res := db.Cmd("ZRANK", key, client.ID)
-		assert.Equal(t, redis.IntegerReply, res.Type, "res: %s", res)
+		assert.Equal(t, true, res.IsType(redis.Int), "res: %s", res)
 	}
 
 	err = UpdateQueues(client, []string{})
@@ -52,7 +52,7 @@ func TestUpdateQueues(t *T) {
 	for i := range queues {
 		key := db.ConsumersKey(queues[i])
 		res := db.Cmd("ZRANK", key, client.ID)
-		assert.Equal(t, redis.NilReply, res.Type, "res: %s", res)
+		assert.Equal(t, true, res.IsType(redis.Nil), "res: %s", res)
 	}
 }
 
@@ -66,7 +66,7 @@ func TestStaleCleanup(t *T) {
 	// Make sure the queue has this clientId as a consumer
 	key := db.ConsumersKey(queue)
 	res := db.Cmd("ZRANK", key, client.ID)
-	assert.Equal(t, redis.IntegerReply, res.Type, "res: %s", res)
+	assert.Equal(t, true, res.IsType(redis.Int), "res: %s", res)
 
 	// Remove all knowledge about this client from the consumer state
 	callCh <- func(s *state) {
@@ -80,5 +80,5 @@ func TestStaleCleanup(t *T) {
 
 	// Make sure this client is no longer a consumer
 	res = db.Cmd("ZRANK", key, client.ID)
-	assert.Equal(t, redis.NilReply, res.Type, "key: %s clientId: %s res: %s", key, client.ID, res)
+	assert.Equal(t, true, res.IsType(redis.Nil), "key: %s clientId: %s res: %s", key, client.ID, res)
 }

commands/commands.go

@@ -9,13 +9,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/fzzy/radix/redis"
-	"github.com/fzzy/radix/redis/resp"
-
 	"github.com/mc0/okq/clients"
 	"github.com/mc0/okq/clients/consumers"
 	"github.com/mc0/okq/db"
 	"github.com/mc0/okq/log"
+	"github.com/mediocregopher/radix.v2/redis"
 )
 
 type commandFunc func(*clients.Client, []string) (interface{}, error)
@@ -38,7 +36,7 @@ var commandMap = map[string]commandInfo{
 	"PING":      {ping, 0},
 }
 
-var okSS = resp.NewSimpleString("OK")
+var okSS = redis.NewRespSimple("OK")
 
 // Dispatch takes in a client whose command has already been read off the
 // socket, a list of arguments from that command (not including the command name
@@ -62,7 +60,7 @@ func Dispatch(client *clients.Client, cmd string, args []string) {
 		return
 	}
 
-	resp.WriteArbitrary(client.Conn, ret)
+	redis.NewResp(ret).WriteTo(client.Conn)
 }
 
 func parseInt(from, as string) (int, error) {
@@ -77,8 +75,8 @@ func parseInt(from, as string) (int, error) {
 
 func drainPipeline(redisClient *redis.Client) error {
 	for {
-		err := redisClient.GetReply().Err
-		if err == redis.PipelineQueueEmptyError {
+		err := redisClient.PipeResp().Err
+		if err == redis.ErrPipelineEmpty {
 			break
 		} else if err != nil {
 			return err
@@ -87,9 +85,9 @@ func drainPipeline(redisClient *redis.Client) error {
 	return nil
 }
 
-func writeErrf(w io.Writer, format string, args ...interface{}) error {
+func writeErrf(w io.Writer, format string, args ...interface{}) {
 	err := fmt.Errorf(format, args...)
-	return resp.WriteArbitrary(w, err)
+	redis.NewResp(err).WriteTo(w)
 }
 
 func qregister(client *clients.Client, args []string) (interface{}, error) {
@@ -134,7 +132,7 @@ func qpeekgeneric(
 
 	var eventRaw string
 	reply := db.Cmd("HGET", itemsKey, eventID)
-	if reply.Type == redis.NilReply {
+	if reply.IsType(redis.Nil) {
 		return nil, nil
 	}
 	if eventRaw, err = reply.Str(); err != nil {
@@ -164,7 +162,7 @@ func qrpop(client *clients.Client, args []string) (interface{}, error) {
 	unclaimedKey := db.UnclaimedKey(queueName)
 	claimedKey := db.ClaimedKey(queueName)
 	reply := db.Cmd("RPOPLPUSH", unclaimedKey, claimedKey)
-	if reply.Type == redis.NilReply {
+	if reply.IsType(redis.Nil) {
 		return nil, nil
 	}
 
@@ -365,8 +363,8 @@ func qstatus(client *clients.Client, args []string) (interface{}, error) {
 	return queueStatuses, nil
 }
 
-var pong = resp.NewSimpleString("PONG")
+var pongSS = redis.NewRespSimple("PONG")
 
 func ping(client *clients.Client, args []string) (interface{}, error) {
-	return pong, nil
+	return pongSS, nil
 }

commands/commands_test.go

@@ -6,38 +6,42 @@ import (
 	. "testing"
 	"time"
 
-	"github.com/fzzy/radix/redis/resp"
+	"github.com/mediocregopher/radix.v2/redis"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/mc0/okq/clients"
 )
 
 func readAndAssertStr(t *T, client *clients.Client, expected string) {
-	m, err := resp.ReadMessage(client.Conn)
-	require.Nil(t, err, "stack:\n%s", debug.Stack())
+	rr := redis.NewRespReader(client.Conn)
+	m := rr.Read()
+	require.Nil(t, m.Err, "stack:\n%s", debug.Stack())
 	s, err := m.Str()
 	require.Nil(t, err, "stack:\n%s", debug.Stack())
 	assert.Equal(t, expected, s, "m: %v stack:\n%s", m, debug.Stack())
 }
 
 func readAndAssertInt(t *T, client *clients.Client, expected int64) {
-	m, err := resp.ReadMessage(client.Conn)
-	require.Nil(t, err, "stack:\n%s", debug.Stack())
+	rr := redis.NewRespReader(client.Conn)
+	m := rr.Read()
+	require.Nil(t, m.Err, "stack:\n%s", debug.Stack())
 	i, err := m.Int()
 	require.Nil(t, err, "stack:\n%s", debug.Stack())
 	assert.Equal(t, expected, i, "m: %v stack:\n%s", m, debug.Stack())
 }
 
 func readAndAssertNil(t *T, client *clients.Client) {
-	m, err := resp.ReadMessage(client.Conn)
-	require.Nil(t, err, "stack:\n%s", debug.Stack())
-	assert.Equal(t, resp.Nil, m.Type, "m: %v stack:\n%s", m, debug.Stack())
+	rr := redis.NewRespReader(client.Conn)
+	m := rr.Read()
+	require.Nil(t, m.Err, "stack:\n%s", debug.Stack())
+	assert.Equal(t, true, m.IsType(redis.Nil), "m: %v stack:\n%s", m, debug.Stack())
 }
 
 func readAndAssertArr(t *T, client *clients.Client, expected []string) {
-	m, err := resp.ReadMessage(client.Conn)
-	require.Nil(t, err, "stack:\n%s", debug.Stack())
+	rr := redis.NewRespReader(client.Conn)
+	m := rr.Read()
+	require.Nil(t, m.Err, "stack:\n%s", debug.Stack())
 
 	arr, err := m.Array()
 	require.Nil(t, err, "stack:\n%s", debug.Stack())

config/config.go

@@ -3,9 +3,7 @@
 package config
 
 import (
-	"log" // we don't use our log because it imports this package
-
-	"github.com/mediocregopher/flagconfig"
+	"github.com/mediocregopher/lever"
 )
 
 // Variables populated by the flag parsing process at runtime
@@ -17,19 +15,32 @@ var (
 )
 
 func init() {
-	fc := flagconfig.New("redeqeue")
-
-	fc.StrParam("listen-addr", "Address to listen for client connections on", ":4777")
-	fc.StrParam("redis-addr", "Address redis is listening on", "127.0.0.1:6379")
-	fc.FlagParam("redis-cluster", "Whether or not to treat the redis address as a node in a larger cluster", false)
-	fc.FlagParam("debug", "Turn on debug logging", false)
-
-	if err := fc.Parse(); err != nil {
-		log.Fatal(err)
-	}
+	l := lever.New("okq", nil)
+	l.Add(lever.Param{
+		Name:        "--listen-addr",
+		Description: "Address to listen for client connections on",
+		Default:     ":4777",
+	})
+	l.Add(lever.Param{
+		Name:        "--redis-addr",
+		Description: "Address redis is listening on",
+		Default:     "127.0.0.1:6379",
+	})
+	l.Add(lever.Param{
+		Name:        "--redis-cluster",
+		Description: "Whether or not to treat the redis address as a node in a larger cluster",
+		Flag:        true,
+	})
+	l.Add(lever.Param{
+		Name:        "--debug",
+		Aliases:     []string{"-d"},
+		Description: "Turn on debug logging",
+		Flag:        true,
+	})
+	l.Parse()
 
-	ListenAddr = fc.GetStr("listen-addr")
-	RedisAddr = fc.GetStr("redis-addr")
-	RedisCluster = fc.GetFlag("redis-cluster")
-	Debug = fc.GetFlag("debug")
+	ListenAddr, _ = l.ParamStr("--listen-addr")
+	RedisAddr, _ = l.ParamStr("--redis-addr")
+	RedisCluster = l.ParamFlag("--redis-cluster")
+	Debug = l.ParamFlag("debug")
 }

db/db.go

@@ -6,28 +6,28 @@ import (
 	"errors"
 	"strings"
 
-	"github.com/fzzy/radix/redis"
 	"github.com/mc0/okq/config"
 	"github.com/mc0/okq/log"
+	"github.com/mediocregopher/radix.v2/redis"
 )
 
 // These functions are filled in dynamically at runtime
 var (
 	// Cmd is a function which will perform the given cmd/args in redis and
-	// returns the reply. It automatically handles using redis cluster, if that
+	// returns the resp. It automatically handles using redis cluster, if that
 	// is enabled
-	Cmd func(string, ...interface{}) *redis.Reply
+	Cmd func(string, ...interface{}) *redis.Resp
 
 	// Pipe runs a set of commands (given by p) one after the other. It is *not*
 	// guaranteed that all the commands will be run on the same client. If any
 	// commands return an error the pipeline will stop and return that error.
-	// Otherwise the Reply from each command is returned in a slice
+	// Otherwise the Resp from each command is returned in a slice
 	//
 	//	r, err := db.Pipe(
 	//		db.PP("SET", "foo", "bar"),
 	//		db.PP("GET", "foo"),
 	//	)
-	Pipe func(...*PipePart) ([]*redis.Reply, error)
+	Pipe func(...*PipePart) ([]*redis.Resp, error)
 
 	// Scan is a function which returns a channel to which keys matching the
 	// given pattern are written to. The channel must be read from until it is
@@ -36,6 +36,16 @@ var (
 	//
 	// This should not be used in any critical paths
 	Scan func(string) <-chan string
+
+	// Lua performs one of the preloaded Lua scripts that have been built-in.
+	// It's *possible* that the script wasn't loaded in initLuaScripts() for
+	// some strange reason, this tries to handle that case as well. The integer
+	// passed in is the number of keys the command takes in
+	//
+	// Example:
+	//
+	//	db.Lua(redisClient, "LREMRPUSH", 2, "foo", "bar", "value")
+	Lua func(string, int, ...interface{}) *redis.Resp
 )
 
 // PipePart is a single command to be run in a pipe. See Pipe for an example on
@@ -147,8 +157,12 @@ func scanHelper(redisClient *redis.Client, pattern string, retCh chan string) er
 		if r.Err != nil {
 			return r.Err
 		}
+		elems, err := r.Array()
+		if err != nil {
+			return err
+		}
 
-		results, err := r.Elems[1].List()
+		results, err := elems[1].List()
 		if err != nil {
 			return err
 		}
@@ -157,7 +171,7 @@ func scanHelper(redisClient *redis.Client, pattern string, retCh chan string) er
 			retCh <- results[i]
 		}
 
-		if cursor, err = r.Elems[0].Str(); err != nil {
+		if cursor, err = elems[0].Str(); err != nil {
 			return err
 		} else if cursor == "0" {
 			return nil