parser: greatly simplify reply parsing.

commands.go

@@ -42,20 +42,6 @@ func (c *Client) Select(index int64) *StatusReq {
 
 //------------------------------------------------------------------------------
 
-func (c *Client) Flushall() *StatusReq {
-	req := NewStatusReq("FLUSHALL")
-	c.Process(req)
-	return req
-}
-
-func (c *Client) Flushdb() *StatusReq {
-	req := NewStatusReq("FLUSHDB")
-	c.Process(req)
-	return req
-}
-
-//------------------------------------------------------------------------------
-
 func (c *Client) Del(keys ...string) *IntReq {
 	args := append([]string{"DEL"}, keys...)
 	req := NewIntReq(args...)
@@ -746,8 +732,8 @@ func (c *Client) ZRangeByScore(
 	return req
 }
 
-func (c *Client) ZRank(key, member string) *IntNilReq {
-	req := NewIntNilReq("ZRANK", key, member)
+func (c *Client) ZRank(key, member string) *IntReq {
+	req := NewIntReq("ZRANK", key, member)
 	c.Process(req)
 	return req
 }
@@ -808,8 +794,8 @@ func (c *Client) ZRevRangeByScore(
 	return req
 }
 
-func (c *Client) ZRevRank(key, member string) *IntNilReq {
-	req := NewIntNilReq("ZREVRANK", key, member)
+func (c *Client) ZRevRank(key, member string) *IntReq {
+	req := NewIntReq("ZREVRANK", key, member)
 	c.Process(req)
 	return req
 }
@@ -842,3 +828,111 @@ func (c *Client) ZUnionStore(
 	c.Process(req)
 	return req
 }
+
+//------------------------------------------------------------------------------
+
+func (c *Client) BgRewriteAOF() *StatusReq {
+	req := NewStatusReq("BGREWRITEAOF")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) BgSave() *StatusReq {
+	req := NewStatusReq("BGSAVE")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) ClientKill(ipPort string) *StatusReq {
+	req := NewStatusReq("CLIENT", "KILL", ipPort)
+	c.Process(req)
+	return req
+}
+
+func (c *Client) ClientList() *BulkReq {
+	req := NewBulkReq("CLIENT", "LIST")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) ConfigGet(parameter string) *MultiBulkReq {
+	req := NewMultiBulkReq("CONFIG", "GET", parameter)
+	c.Process(req)
+	return req
+}
+
+func (c *Client) ConfigResetStat() *StatusReq {
+	req := NewStatusReq("CONFIG", "RESETSTAT")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) ConfigSet(parameter, value string) *StatusReq {
+	req := NewStatusReq("CONFIG", "SET", parameter, value)
+	c.Process(req)
+	return req
+}
+
+func (c *Client) DbSize() *IntReq {
+	req := NewIntReq("DBSIZE")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) FlushAll() *StatusReq {
+	req := NewStatusReq("FLUSHALL")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) FlushDb() *StatusReq {
+	req := NewStatusReq("FLUSHDB")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) Info() *BulkReq {
+	req := NewBulkReq("INFO")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) LastSave() *IntReq {
+	req := NewIntReq("LASTSAVE")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) Monitor() {
+	panic("not implemented")
+}
+
+func (c *Client) Save() *StatusReq {
+	req := NewStatusReq("SAVE")
+	c.Process(req)
+	return req
+}
+
+func (c *Client) Shutdown() {
+	panic("not implemented")
+}
+
+func (c *Client) SlaveOf(host, port string) *StatusReq {
+	req := NewStatusReq("SLAVEOF", host, port)
+	c.Process(req)
+	return req
+}
+
+func (c *Client) SlowLog() {
+	panic("not implemented")
+}
+
+func (c *Client) Sync() {
+	panic("not implemented")
+}
+
+func (c *Client) Time() *MultiBulkReq {
+	req := NewMultiBulkReq("TIME")
+	c.Process(req)
+	return req
+}

connpool.go

@@ -1,11 +1,12 @@
 package redis
 
 import (
-	"bufio"
 	"io"
 	"log"
 	"os"
 	"sync"
+
+	"github.com/vmihailenco/bufio"
 )
 
 type Conn struct {

multi.go

@@ -105,7 +105,7 @@ func (c *MultiClient) ExecReqs(reqs []Req, conn *Conn) error {
 	if line[0] != '*' {
 		return fmt.Errorf("Expected '*', but got line %q", line)
 	}
-	if isNilReplies(line) {
+	if len(line) == 3 && line[1] == '-' && line[2] == '1' {
 		return Nil
 	}
 

parser.go

@@ -0,0 +1,170 @@
+package redis
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+
+	"github.com/vmihailenco/bufio"
+)
+
+//------------------------------------------------------------------------------
+
+var Nil = errors.New("(nil)")
+
+//------------------------------------------------------------------------------
+
+var (
+	errReaderTooSmall = errors.New("redis: reader is too small")
+	errValNotSet      = errors.New("redis: value is not set")
+)
+
+//------------------------------------------------------------------------------
+
+func PackReq(args []string) []byte {
+	buf := make([]byte, 0, 1024)
+	buf = append(buf, '*')
+	buf = strconv.AppendUint(buf, uint64(len(args)), 10)
+	buf = append(buf, '\r', '\n')
+	for _, arg := range args {
+		buf = append(buf, '$')
+		buf = strconv.AppendUint(buf, uint64(len(arg)), 10)
+		buf = append(buf, '\r', '\n')
+		buf = append(buf, []byte(arg)...)
+		buf = append(buf, '\r', '\n')
+	}
+	return buf
+}
+
+//------------------------------------------------------------------------------
+
+type ReadLiner interface {
+	ReadLine() ([]byte, bool, error)
+	Peek(n int) ([]byte, error)
+	ReadN(n int) ([]byte, error)
+}
+
+func readLine(rd ReadLiner) ([]byte, error) {
+	line, isPrefix, err := rd.ReadLine()
+	if err != nil {
+		return line, err
+	}
+	if isPrefix {
+		return line, errReaderTooSmall
+	}
+	return line, nil
+}
+
+//------------------------------------------------------------------------------
+
+func ParseReq(rd ReadLiner) ([]string, error) {
+	line, err := readLine(rd)
+	if err != nil {
+		return nil, err
+	}
+
+	if line[0] != '*' {
+		return []string{string(line)}, nil
+	}
+	numReplies, err := strconv.ParseInt(string(line[1:]), 10, 64)
+	if err != nil {
+		return nil, err
+	}
+
+	args := make([]string, 0)
+	for i := int64(0); i < numReplies; i++ {
+		line, err = readLine(rd)
+		if err != nil {
+			return nil, err
+		}
+		if line[0] != '$' {
+			return nil, fmt.Errorf("Expected '$', but got %q", line)
+		}
+
+		line, err = readLine(rd)
+		if err != nil {
+			return nil, err
+		}
+		args = append(args, string(line))
+	}
+	return args, nil
+}
+
+//------------------------------------------------------------------------------
+
+func ParseReply(rd ReadLiner) (interface{}, error) {
+	line, err := readLine(rd)
+	if err != nil {
+		return 0, err
+	}
+
+	switch line[0] {
+	case '-':
+		return nil, errors.New(string(line[1:]))
+	case '+':
+		return string(line[1:]), nil
+	case ':':
+		return strconv.ParseInt(string(line[1:]), 10, 64)
+	case '$':
+		if len(line) == 3 && line[1] == '-' && line[2] == '1' {
+			return "", Nil
+		} else if len(line) == 2 && line[1] == '0' {
+			return "", nil
+		}
+
+		replyLenInt32, err := strconv.ParseInt(string(line[1:]), 10, 32)
+		if err != nil {
+			return "", err
+		}
+		replyLen := int(replyLenInt32) + 2
+
+		line, err = rd.ReadN(replyLen)
+		if err == bufio.ErrBufferFull {
+			buf := make([]byte, replyLen)
+			r := 0
+
+			r += copy(buf, line)
+
+			for err == bufio.ErrBufferFull {
+				line, err = rd.ReadN(replyLen - r)
+				r += copy(buf[r:], line)
+			}
+
+			line = buf
+		} else if err != nil {
+			return "", err
+		}
+
+		return string(line[:len(line)-2]), nil
+	case '*':
+		if len(line) == 3 && line[1] == '-' && line[2] == '1' {
+			return nil, Nil
+		}
+
+		val := make([]interface{}, 0)
+		if len(line) == 2 && line[1] == '0' {
+			return val, nil
+		}
+
+		numReplies, err := strconv.ParseInt(string(line[1:]), 10, 64)
+		if err != nil {
+			return nil, err
+		}
+
+		for i := int64(0); i < numReplies; i++ {
+			v, err := ParseReply(rd)
+			if err == Nil {
+				val = append(val, nil)
+			} else if err != nil {
+				return nil, err
+			} else {
+				val = append(val, v)
+			}
+		}
+
+		return val, nil
+	default:
+		return nil, fmt.Errorf("redis: can't parse %q", line)
+	}
+	panic("not reachable")
+}

redis.go

@@ -2,16 +2,11 @@ package redis
 
 import (
 	"crypto/tls"
-	"errors"
 	"io"
 	"net"
 	"sync"
 )
 
-var (
-	ErrReaderTooSmall = errors.New("redis: Reader is too small")
-)
-
 type OpenConnFunc func() (io.ReadWriteCloser, error)
 type CloseConnFunc func(io.ReadWriteCloser) error
 type InitConnFunc func(*Client) error
@@ -110,7 +105,11 @@ func (c *BaseClient) Run(req Req) {
 
 	val, err := req.ParseReply(conn.Rd)
 	if err != nil {
+		if err == Nil {
 			c.ConnPool.Add(conn)
+		} else {
+			c.ConnPool.Remove(conn)
+		}
 		req.SetErr(err)
 		return
 	}

redis_test.go

@@ -1,6 +1,7 @@
 package redis_test
 
 import (
+	"bytes"
 	"io"
 	"net"
 	"strconv"
@@ -40,11 +41,11 @@ func (t *RedisTest) SetUpTest(c *C) {
 		return nil
 	}
 	t.client = redis.NewClient(openConn, closeConn, nil)
-	c.Assert(t.client.Flushdb().Err(), IsNil)
+	c.Assert(t.client.FlushDb().Err(), IsNil)
 }
 
 func (t *RedisTest) TearDownTest(c *C) {
-	c.Assert(t.client.Flushdb().Err(), IsNil)
+	c.Assert(t.client.FlushDb().Err(), IsNil)
 	c.Assert(t.client.Close(), IsNil)
 	c.Assert(t.openedConnsCount, Equals, t.closedConnsCount)
 }
@@ -75,6 +76,8 @@ func (t *RedisTest) TestInitConn(c *C) {
 
 func (t *RedisTest) TestRunWithouthCheckingErrVal(c *C) {
 	set := t.client.Set("foo", "bar")
+	c.Assert(set.Err(), IsNil)
+	c.Assert(set.Val(), Equals, "OK")
 
 	get := t.client.Get("foo")
 	c.Assert(get.Err(), IsNil)
@@ -84,6 +87,28 @@ func (t *RedisTest) TestRunWithouthCheckingErrVal(c *C) {
 	c.Assert(set.Val(), Equals, "OK")
 }
 
+func (t *RedisTest) TestGetSpecChars(c *C) {
+	set := t.client.Set("foo", "bar1\r\nbar2\r\n")
+	c.Assert(set.Err(), IsNil)
+	c.Assert(set.Val(), Equals, "OK")
+
+	get := t.client.Get("foo")
+	c.Assert(get.Err(), IsNil)
+	c.Assert(get.Val(), Equals, "bar1\r\nbar2\r\n")
+}
+
+func (t *RedisTest) TestGetBigVal(c *C) {
+	val := string(bytes.Repeat([]byte{'*'}, 2<<16))
+
+	set := t.client.Set("foo", val)
+	c.Assert(set.Err(), IsNil)
+	c.Assert(set.Val(), Equals, "OK")
+
+	get := t.client.Get("foo")
+	c.Assert(get.Err(), IsNil)
+	c.Assert(get.Val(), Equals, val)
+}
+
 //------------------------------------------------------------------------------
 
 func (t *RedisTest) TestConnPoolMaxCap(c *C) {
@@ -191,7 +216,7 @@ func (t *RedisTest) TestConnPoolRemovesBrokenConn(c *C) {
 	client := redis.NewTCPClient(redisAddr, "", -1)
 	client.ConnPool.(*redis.MultiConnPool).MaxCap = 1
 	defer func() {
-		c.Check(client.Close(), IsNil)
+		c.Assert(client.Close(), IsNil)
 	}()
 
 	c.Assert(client.ConnPool.Add(redis.NewConn(conn)), IsNil)
@@ -1919,7 +1944,7 @@ func (t *RedisTest) TestPipelineErrValNotSet(c *C) {
 	}()
 
 	get := pipeline.Get("foo")
-	c.Check(get.Err(), Equals, redis.ErrValNotSet)
+	c.Assert(get.Err(), ErrorMatches, "redis: value is not set")
 }
 
 func (t *RedisTest) TestPipelineRunQueuedOnEmptyQueue(c *C) {
@@ -2053,7 +2078,7 @@ func (t *RedisTest) TestMultiExecOnEmptyQueue(c *C) {
 	multi, err := t.client.MultiClient()
 	c.Assert(err, IsNil)
 	defer func() {
-		c.Check(multi.Close(), IsNil)
+		c.Assert(multi.Close(), IsNil)
 	}()
 
 	reqs, err := multi.Exec()
@@ -2063,7 +2088,7 @@ func (t *RedisTest) TestMultiExecOnEmptyQueue(c *C) {
 
 //------------------------------------------------------------------------------
 
-func (t *RedisTest) TestEchoFromGoroutines(c *C) {
+func (t *RedisTest) TestSyncEchoFromGoroutines(c *C) {
 	wg := &sync.WaitGroup{}
 	for i := int64(0); i < 1000; i++ {
 		wg.Add(1)
@@ -2181,6 +2206,109 @@ func (t *RedisTest) TestWatchUnwatch(c *C) {
 
 //------------------------------------------------------------------------------
 
+func (t *RedisTest) TestCmdBgRewriteAOF(c *C) {
+	r := t.client.BgRewriteAOF()
+	c.Assert(r.Err(), IsNil)
+	c.Assert(r.Val(), Equals, "Background append only file rewriting started")
+}
+
+func (t *RedisTest) TestCmdBgSave(c *C) {
+	r := t.client.BgSave()
+	c.Assert(r.Err(), ErrorMatches, "ERR Can't BGSAVE while AOF log rewriting is in progress")
+	c.Assert(r.Val(), Equals, "")
+}
+
+func (t *RedisTest) TestCmdClientKill(c *C) {
+	r := t.client.ClientKill("1.1.1.1:1111")
+	c.Assert(r.Err(), ErrorMatches, "ERR No such client")
+	c.Assert(r.Val(), Equals, "")
+}
+
+func (t *RedisTest) TestCmdClientList(c *C) {
+	r := t.client.ClientList()
+	c.Assert(r.Err(), IsNil)
+	c.Assert(
+		r.Val(),
+		Matches,
+		"addr=127.0.0.1:[0-9]+ fd=[0-9]+ idle=0 flags=N db=0 sub=0 psub=0 qbuf=0 obl=0 oll=0 events=r cmd=client\n",
+	)
+}
+
+func (t *RedisTest) TestCmdConfigGet(c *C) {
+	r := t.client.ConfigGet("*")
+	c.Assert(r.Err(), IsNil)
+	c.Assert(len(r.Val()) > 0, Equals, true)
+}
+
+func (t *RedisTest) TestCmdConfigResetStat(c *C) {
+	r := t.client.ConfigResetStat()
+	c.Assert(r.Err(), IsNil)
+	c.Assert(r.Val(), Equals, "OK")
+}
+
+func (t *RedisTest) TestCmdConfigSet(c *C) {
+	configGet := t.client.ConfigGet("maxmemory")
+	c.Assert(configGet.Err(), IsNil)
+	c.Assert(configGet.Val(), HasLen, 2)
+	c.Assert(configGet.Val()[0].(string), Equals, "maxmemory")
+
+	configSet := t.client.ConfigSet("maxmemory", configGet.Val()[1].(string))
+	c.Assert(configSet.Err(), IsNil)
+	c.Assert(configSet.Val(), Equals, "OK")
+}
+
+func (t *RedisTest) TestCmdDbSize(c *C) {
+	dbSize := t.client.DbSize()
+	c.Assert(dbSize.Err(), IsNil)
+	c.Assert(dbSize.Val(), Equals, int64(0))
+}
+
+func (t *RedisTest) TestCmdFlushAll(c *C) {
+	// TODO
+}
+
+func (t *RedisTest) TestCmdFlushDb(c *C) {
+	// TODO
+}
+
+func (t *RedisTest) TestCmdInfo(c *C) {
+	info := t.client.Info()
+	c.Check(info.Err(), IsNil)
+	c.Check(info.Val(), Not(Equals), "")
+}
+
+func (t *RedisTest) TestCmdLastSave(c *C) {
+	lastSave := t.client.LastSave()
+	c.Check(lastSave.Err(), IsNil)
+	c.Check(lastSave.Val(), Not(Equals), 0)
+}
+
+func (t *RedisTest) TestCmdSave(c *C) {
+	save := t.client.Save()
+	c.Check(save.Err(), IsNil)
+	c.Check(save.Val(), Equals, "OK")
+}
+
+func (t *RedisTest) TestSlaveOf(c *C) {
+	slaveOf := t.client.SlaveOf("localhost", "8888")
+	c.Check(slaveOf.Err(), IsNil)
+	c.Check(slaveOf.Val(), Equals, "OK")
+
+	slaveOf = t.client.SlaveOf("NO", "ONE")
+	c.Check(slaveOf.Err(), IsNil)
+	c.Check(slaveOf.Val(), Equals, "OK")
+}
+
+func (t *RedisTest) TestTime(c *C) {
+	c.Skip("2.6")
+
+	time := t.client.Time()
+	c.Check(time.Err(), IsNil)
+	c.Check(time.Val(), HasLen, 2)
+}
+
+//------------------------------------------------------------------------------
+
 func (t *RedisTest) BenchmarkRedisPing(c *C) {
 	c.StopTimer()
 

request.go

@@ -1,104 +1,9 @@
 package redis
 
 import (
-	"errors"
-	"fmt"
 	"strconv"
 )
 
-var Nil = errors.New("(nil)")
-
-var ErrValNotSet = errors.New("redis: value is not set")
-
-//------------------------------------------------------------------------------
-
-func isNil(line []byte) bool {
-	return len(line) == 3 && line[0] == '$' && line[1] == '-' && line[2] == '1'
-}
-
-func isEmpty(line []byte) bool {
-	return len(line) == 2 && line[0] == '$' && line[1] == '0'
-}
-
-func isNilReplies(line []byte) bool {
-	return len(line) == 3 && line[0] == '*' && line[1] == '-' && line[2] == '1'
-}
-
-func isNoReplies(line []byte) bool {
-	return len(line) == 2 && line[1] == '*' && line[1] == '0'
-}
-
-//------------------------------------------------------------------------------
-
-type ReadLiner interface {
-	ReadLine() ([]byte, bool, error)
-}
-
-func readLine(rd ReadLiner) ([]byte, error) {
-	line, isPrefix, err := rd.ReadLine()
-	if err != nil {
-		return line, err
-	}
-	if isPrefix {
-		return line, ErrReaderTooSmall
-	}
-	return line, nil
-}
-
-//------------------------------------------------------------------------------
-
-func ParseReq(rd ReadLiner) ([]string, error) {
-	line, err := readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	if line[0] != '*' {
-		return []string{string(line)}, nil
-	}
-	numReplies, err := strconv.ParseInt(string(line[1:]), 10, 64)
-	if err != nil {
-		return nil, err
-	}
-
-	args := make([]string, 0)
-	for i := int64(0); i < numReplies; i++ {
-		line, err = readLine(rd)
-		if err != nil {
-			return nil, err
-		}
-		if line[0] != '$' {
-			return nil, fmt.Errorf("Expected '$', but got %q", line)
-		}
-
-		line, err = readLine(rd)
-		if err != nil {
-			return nil, err
-		}
-		args = append(args, string(line))
-	}
-	return args, nil
-}
-
-//------------------------------------------------------------------------------
-
-func PackReq(args []string) []byte {
-	buf := make([]byte, 0, 1024)
-	buf = append(buf, '*')
-	buf = strconv.AppendUint(buf, uint64(len(args)), 10)
-	buf = append(buf, '\r', '\n')
-	for _, arg := range args {
-		buf = append(buf, '$')
-		buf = strconv.AppendUint(buf, uint64(len(arg)), 10)
-		buf = append(buf, '\r', '\n')
-		buf = append(buf, []byte(arg)...)
-		buf = append(buf, '\r', '\n')
-	}
-	return buf
-}
-
-//------------------------------------------------------------------------------
-
 type Req interface {
 	Req() []byte
 	ParseReply(ReadLiner) (interface{}, error)
@@ -139,7 +44,7 @@ func (r *BaseReq) Err() error {
 		return r.err
 	}
 	if r.val == nil {
-		return ErrValNotSet
+		return errValNotSet
 	}
 	return nil
 }
@@ -156,7 +61,7 @@ func (r *BaseReq) InterfaceVal() interface{} {
 }
 
 func (r *BaseReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	panic("abstract")
+	return ParseReply(rd)
 }
 
 //------------------------------------------------------------------------------
@@ -171,21 +76,6 @@ func NewStatusReq(args ...string) *StatusReq {
 	}
 }
 
-func (r *StatusReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] != '+' {
-		return nil, fmt.Errorf("Expected '+', but got %q", line)
-	}
-
-	return string(line[1:]), nil
-}
-
 func (r *StatusReq) Val() string {
 	if r.val == nil {
 		return ""
@@ -205,21 +95,6 @@ func NewIntReq(args ...string) *IntReq {
 	}
 }
 
-func (r *IntReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] != ':' {
-		return nil, fmt.Errorf("Expected ':', but got line %q", line)
-	}
-
-	return strconv.ParseInt(string(line[1:]), 10, 64)
-}
-
 func (r *IntReq) Val() int64 {
 	if r.val == nil {
 		return 0
@@ -229,42 +104,6 @@ func (r *IntReq) Val() int64 {
 
 //------------------------------------------------------------------------------
 
-type IntNilReq struct {
-	*BaseReq
-}
-
-func NewIntNilReq(args ...string) *IntNilReq {
-	return &IntNilReq{
-		BaseReq: NewBaseReq(args...),
-	}
-}
-
-func (r *IntNilReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] == ':' {
-		return strconv.ParseInt(string(line[1:]), 10, 64)
-	} else if isNil(line) {
-		return nil, Nil
-	}
-
-	return nil, fmt.Errorf("Expected ':', but got line %q", line)
-}
-
-func (r *IntNilReq) Val() int64 {
-	if r.val == nil {
-		return 0
-	}
-	return r.val.(int64)
-}
-
-//------------------------------------------------------------------------------
-
 type BoolReq struct {
 	*BaseReq
 }
@@ -276,18 +115,11 @@ func NewBoolReq(args ...string) *BoolReq {
 }
 
 func (r *BoolReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
+	v, err := ParseReply(rd)
 	if err != nil {
 		return nil, err
 	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] != ':' {
-		return nil, fmt.Errorf("Expected ':', but got line %q", line)
-	}
-
-	return line[1] == '1', nil
+	return v.(int64) == 1, nil
 }
 
 func (r *BoolReq) Val() bool {
@@ -309,30 +141,6 @@ func NewBulkReq(args ...string) *BulkReq {
 	}
 }
 
-func (r *BulkReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] != '$' {
-		return nil, fmt.Errorf("Expected '$', but got line %q", line)
-	}
-
-	if isNil(line) {
-		return nil, Nil
-	}
-
-	line, err = readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	return string(line), nil
-}
-
 func (r *BulkReq) Val() string {
 	if r.val == nil {
 		return ""
@@ -353,27 +161,11 @@ func NewFloatReq(args ...string) *FloatReq {
 }
 
 func (r *FloatReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
+	v, err := ParseReply(rd)
 	if err != nil {
 		return nil, err
 	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] != '$' {
-		return nil, fmt.Errorf("Expected '$', but got line %q", line)
-	}
-
-	if isNil(line) {
-		return nil, Nil
-	}
-
-	line, err = readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	return strconv.ParseFloat(string(line), 64)
+	return strconv.ParseFloat(v.(string), 64)
 }
 
 func (r *FloatReq) Val() float64 {
@@ -395,63 +187,6 @@ func NewMultiBulkReq(args ...string) *MultiBulkReq {
 	}
 }
 
-func (r *MultiBulkReq) ParseReply(rd ReadLiner) (interface{}, error) {
-	line, err := readLine(rd)
-	if err != nil {
-		return nil, err
-	}
-
-	if line[0] == '-' {
-		return nil, errors.New(string(line[1:]))
-	} else if line[0] != '*' {
-		return nil, fmt.Errorf("Expected '*', but got line %q", line)
-	} else if isNilReplies(line) {
-		return nil, Nil
-	}
-
-	val := make([]interface{}, 0)
-	if isNoReplies(line) {
-		return val, nil
-	}
-	numReplies, err := strconv.ParseInt(string(line[1:]), 10, 64)
-	if err != nil {
-		return nil, err
-	}
-
-	for i := int64(0); i < numReplies; i++ {
-		line, err = readLine(rd)
-		if err != nil {
-			return nil, err
-		}
-
-		switch line[0] {
-		case ':':
-			var n int64
-			n, err = strconv.ParseInt(string(line[1:]), 10, 64)
-			if err != nil {
-				return nil, err
-			}
-			val = append(val, n)
-		case '$':
-			if isEmpty(line) {
-				val = append(val, "")
-			} else if isNil(line) {
-				val = append(val, nil)
-			} else {
-				line, err = readLine(rd)
-				if err != nil {
-					return nil, err
-				}
-				val = append(val, string(line))
-			}
-		default:
-			return nil, fmt.Errorf("Expected '$', but got line %q", line)
-		}
-	}
-
-	return val, nil
-}
-
 func (r *MultiBulkReq) Val() []interface{} {
 	if r.val == nil {
 		return nil

request_test.go

@@ -1,8 +1,7 @@
 package redis_test
 
 import (
-	"bufio"
-
+	"github.com/vmihailenco/bufio"
 	. "launchpad.net/gocheck"
 
 	"github.com/vmihailenco/redis"