From 41137c2e6ff7d88c1b41c71e28714fa058500a01 Mon Sep 17 00:00:00 2001 From: Vladimir Mihailenco Date: Sun, 29 Jul 2012 12:42:00 +0300 Subject: [PATCH] Add pipelining support. --- README.md | 44 ++++++++++- commands.go | 214 +++++++++++++++++++++++++------------------------- redis.go | 71 ++++++++++++----- redis_test.go | 27 ++++++- request.go | 86 +++++++++++++++++++- 5 files changed, 310 insertions(+), 132 deletions(-) diff --git a/README.md b/README.md index 93a7081..ddd4a4b 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,31 @@ Example: } } +Pipelining +---------- + +Client has ability to run several commands with one read/write: + + setReq := redisClient.Set("foo1", "bar1") // queue command SET + getReq := redisClient.Get("foo2") // queue command GET + + reqs, err := redisClient.RunQueued() // run queued commands + if err != nil { + panic(err) + } + + ok, err := setReq.Reply() + if err != nil { + panic(err) + } + + value, err := getReq.Reply() + if err != nil { + if err != redis.Nil { + panic(err) + } + } + Multi/Exec ---------- @@ -126,15 +151,30 @@ Client is thread safe. Internally sync.Mutex is used to synchronize writes and r Custom commands --------------- -Example: +Lazy command: func Get(client *redis.Client, key string) *redis.BulkReq { req := redis.NewBulkReq("GET", key) - client.Run(req) + client.Queue(req) return req } value, err := Get(redisClient, "foo").Reply() + if err != nil { + if err != redis.Nil { + panic(err) + } + } + +Immediate command: + + func Quit(client *redis.Client) *redis.StatusReq { + req := redis.NewStatusReq("QUIT") + client.Run(req) + return req + } + + status, err := Quit(redisClient).Reply() if err != nil { panic(err) } diff --git a/commands.go b/commands.go index 03ad9ef..9dc7390 100644 --- a/commands.go +++ b/commands.go @@ -18,19 +18,19 @@ func NewLimit(offset, count int64) *Limit { func (c *Client) Auth(password string) *StatusReq { req := NewStatusReq("AUTH", password) - c.Run(req) + c.Queue(req) return req } func (c *Client) Echo(message string) *BulkReq { req := NewBulkReq("ECHO", message) - c.Run(req) + c.Queue(req) return req } func (c *Client) Ping() *StatusReq { req := NewStatusReq("PING") - c.Run(req) + c.Queue(req) return req } @@ -43,7 +43,7 @@ func (c *Client) Quit() *StatusReq { func (c *Client) Select(index int64) *StatusReq { req := NewStatusReq("SELECT", strconv.FormatInt(index, 10)) - c.Run(req) + c.Queue(req) return req } @@ -51,13 +51,13 @@ func (c *Client) Select(index int64) *StatusReq { func (c *Client) Flushall() *StatusReq { req := NewStatusReq("FLUSHALL") - c.Run(req) + c.Queue(req) return req } func (c *Client) Flushdb() *StatusReq { req := NewStatusReq("FLUSHDB") - c.Run(req) + c.Queue(req) return req } @@ -66,37 +66,37 @@ func (c *Client) Flushdb() *StatusReq { func (c *Client) Del(keys ...string) *IntReq { args := append([]string{"DEL"}, keys...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) Dump(key string) *BulkReq { req := NewBulkReq("DUMP", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) Exists(key string) *BoolReq { req := NewBoolReq("EXISTS", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) Expire(key string, seconds int64) *BoolReq { req := NewBoolReq("EXPIRE", key, strconv.FormatInt(seconds, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) ExpireAt(key string, timestamp int64) *BoolReq { req := NewBoolReq("EXPIREAT", key, strconv.FormatInt(timestamp, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) Keys(pattern string) *MultiBulkReq { req := NewMultiBulkReq("KEYS", pattern) - c.Run(req) + c.Queue(req) return req } @@ -109,76 +109,76 @@ func (c *Client) Migrate(host string, port int32, key, db string, timeout int64) db, strconv.FormatInt(timeout, 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) Move(key string, db int64) *BoolReq { req := NewBoolReq("MOVE", key, strconv.FormatInt(db, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) ObjectRefCount(keys ...string) *IntReq { args := append([]string{"OBJECT", "REFCOUNT"}, keys...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) ObjectEncoding(keys ...string) *BulkReq { args := append([]string{"OBJECT", "ENCODING"}, keys...) req := NewBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) ObjectIdleTime(keys ...string) *IntReq { args := append([]string{"OBJECT", "IDLETIME"}, keys...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) Persist(key string) *BoolReq { req := NewBoolReq("PERSIST", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) Pexpire(key string, milliseconds int64) *BoolReq { req := NewBoolReq("PEXPIRE", key, strconv.FormatInt(milliseconds, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) PexpireAt(key string, milliseconds int64) *BoolReq { req := NewBoolReq("PEXPIREAT", key, strconv.FormatInt(milliseconds, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) PTTL(key string) *IntReq { req := NewIntReq("PTTL", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) RandomKey() *BulkReq { req := NewBulkReq("RANDOMKEY") - c.Run(req) + c.Queue(req) return req } func (c *Client) Rename(key, newkey string) *StatusReq { req := NewStatusReq("RENAME", key, newkey) - c.Run(req) + c.Queue(req) return req } func (c *Client) RenameNX(key, newkey string) *BoolReq { req := NewBoolReq("RENAMENX", key, newkey) - c.Run(req) + c.Queue(req) return req } @@ -188,26 +188,26 @@ func (c *Client) Restore(key, ttl int64, value string) *StatusReq { strconv.FormatInt(ttl, 10), value, ) - c.Run(req) + c.Queue(req) return req } func (c *Client) Sort(key string, params ...string) *MultiBulkReq { args := append([]string{"SORT", key}, params...) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) TTL(key string) *IntReq { req := NewIntReq("TTL", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) Type(key string) *StatusReq { req := NewStatusReq("TYPE", key) - c.Run(req) + c.Queue(req) return req } @@ -215,7 +215,7 @@ func (c *Client) Type(key string) *StatusReq { func (c *Client) Append(key, value string) *IntReq { req := NewIntReq("APPEND", key, value) - c.Run(req) + c.Queue(req) return req } @@ -225,25 +225,25 @@ func (c *Client) Append(key, value string) *IntReq { func (c *Client) Decr(key string) *IntReq { req := NewIntReq("DECR", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) DecrBy(key string, decrement int64) *IntReq { req := NewIntReq("DECRBY", key, strconv.FormatInt(decrement, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) Get(key string) *BulkReq { req := NewBulkReq("GET", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) GetBit(key string, offset int64) *IntReq { req := NewIntReq("GETBIT", key, strconv.FormatInt(offset, 10)) - c.Run(req) + c.Queue(req) return req } @@ -254,25 +254,25 @@ func (c *Client) GetRange(key string, start, end int64) *BulkReq { strconv.FormatInt(start, 10), strconv.FormatInt(end, 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) GetSet(key, value string) *BulkReq { req := NewBulkReq("GETSET", key, value) - c.Run(req) + c.Queue(req) return req } func (c *Client) Incr(key string) *IntReq { req := NewIntReq("INCR", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) IncrBy(key string, value int64) *IntReq { req := NewIntReq("INCRBY", key, strconv.FormatInt(value, 10)) - c.Run(req) + c.Queue(req) return req } @@ -281,21 +281,21 @@ func (c *Client) IncrBy(key string, value int64) *IntReq { func (c *Client) MGet(keys ...string) *MultiBulkReq { args := append([]string{"MGET"}, keys...) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) MSet(pairs ...string) *StatusReq { args := append([]string{"MSET"}, pairs...) req := NewStatusReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) MSetNX(pairs ...string) *BoolReq { args := append([]string{"MSETNX"}, pairs...) req := NewBoolReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -306,13 +306,13 @@ func (c *Client) PSetEx(key string, milliseconds int64, value string) *StatusReq strconv.FormatInt(milliseconds, 10), value, ) - c.Run(req) + c.Queue(req) return req } func (c *Client) Set(key, value string) *StatusReq { req := NewStatusReq("SET", key, value) - c.Run(req) + c.Queue(req) return req } @@ -323,31 +323,31 @@ func (c *Client) SetBit(key string, offset int64, value int) *IntReq { strconv.FormatInt(offset, 10), strconv.FormatInt(int64(value), 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) SetEx(key string, seconds int64, value string) *StatusReq { req := NewStatusReq("SETEX", key, strconv.FormatInt(seconds, 10), value) - c.Run(req) + c.Queue(req) return req } func (c *Client) SetNx(key, value string) *BoolReq { req := NewBoolReq("SETNX", key, value) - c.Run(req) + c.Queue(req) return req } func (c *Client) SetRange(key string, offset int64, value string) *IntReq { req := NewIntReq("SETRANGE", key, strconv.FormatInt(offset, 10), value) - c.Run(req) + c.Queue(req) return req } func (c *Client) StrLen(key string) *IntReq { req := NewIntReq("STRLEN", key) - c.Run(req) + c.Queue(req) return req } @@ -356,31 +356,31 @@ func (c *Client) StrLen(key string) *IntReq { func (c *Client) HDel(key string, fields ...string) *IntReq { args := append([]string{"HDEL", key}, fields...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) HExists(key, field string) *BoolReq { req := NewBoolReq("HEXISTS", key, field) - c.Run(req) + c.Queue(req) return req } func (c *Client) HGet(key, field string) *BulkReq { req := NewBulkReq("HGET", key, field) - c.Run(req) + c.Queue(req) return req } func (c *Client) HGetAll(key string) *MultiBulkReq { req := NewMultiBulkReq("HGETALL", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) HIncrBy(key, field string, incr int64) *IntReq { req := NewIntReq("HINCRBY", key, field, strconv.FormatInt(incr, 10)) - c.Run(req) + c.Queue(req) return req } @@ -388,45 +388,45 @@ func (c *Client) HIncrBy(key, field string, incr int64) *IntReq { func (c *Client) HKeys(key string) *MultiBulkReq { req := NewMultiBulkReq("HKEYS", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) HLen(key string) *IntReq { req := NewIntReq("HLEN", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) HMGet(key string, fields ...string) *MultiBulkReq { args := append([]string{"HMGET", key}, fields...) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) HMSet(key, field, value string, pairs ...string) *StatusReq { args := append([]string{"HMSET", key, field, value}, pairs...) req := NewStatusReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) HSet(key, field, value string) *BoolReq { req := NewBoolReq("HSET", key, field, value) - c.Run(req) + c.Queue(req) return req } func (c *Client) HSetNX(key, field, value string) *BoolReq { req := NewBoolReq("HSETNX", key, field, value) - c.Run(req) + c.Queue(req) return req } func (c *Client) HVals(key string) *MultiBulkReq { req := NewMultiBulkReq("HVALS", key) - c.Run(req) + c.Queue(req) return req } @@ -436,7 +436,7 @@ func (c *Client) BLPop(timeout int64, keys ...string) *MultiBulkReq { args := append([]string{"BLPOP"}, keys...) args = append(args, strconv.FormatInt(timeout, 10)) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -444,7 +444,7 @@ func (c *Client) BRPop(timeout int64, keys ...string) *MultiBulkReq { args := append([]string{"BRPOP"}, keys...) args = append(args, strconv.FormatInt(timeout, 10)) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -455,44 +455,44 @@ func (c *Client) BRPopLPush(source, destination string, timeout int64) *BulkReq destination, strconv.FormatInt(timeout, 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) LIndex(key string, index int64) *BulkReq { req := NewBulkReq("LINDEX", key, strconv.FormatInt(index, 10)) - c.Run(req) + c.Queue(req) return req } func (c *Client) LInsert(key, op, pivot, value string) *IntReq { req := NewIntReq("LINSERT", key, op, pivot, value) - c.Run(req) + c.Queue(req) return req } func (c *Client) LLen(key string) *IntReq { req := NewIntReq("LLEN", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) LPop(key string) *BulkReq { req := NewBulkReq("LPOP", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) LPush(key string, values ...string) *IntReq { args := append([]string{"LPUSH", key}, values...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) LPushX(key, value string) *IntReq { req := NewIntReq("LPUSHX", key, value) - c.Run(req) + c.Queue(req) return req } @@ -503,19 +503,19 @@ func (c *Client) LRange(key string, start, stop int64) *MultiBulkReq { strconv.FormatInt(start, 10), strconv.FormatInt(stop, 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) LRem(key string, count int64, value string) *IntReq { req := NewIntReq("LREM", key, strconv.FormatInt(count, 10), value) - c.Run(req) + c.Queue(req) return req } func (c *Client) LSet(key string, index int64, value string) *StatusReq { req := NewStatusReq("LSET", key, strconv.FormatInt(index, 10), value) - c.Run(req) + c.Queue(req) return req } @@ -526,32 +526,32 @@ func (c *Client) LTrim(key string, start, stop int64) *StatusReq { strconv.FormatInt(start, 10), strconv.FormatInt(stop, 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) RPop(key string) *BulkReq { req := NewBulkReq("RPOP", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) RPopLPush(source, destination string) *BulkReq { req := NewBulkReq("RPOPLPUSH", source, destination) - c.Run(req) + c.Queue(req) return req } func (c *Client) RPush(key string, values ...string) *IntReq { args := append([]string{"RPUSH", key}, values...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) RPushX(key string, value string) *IntReq { req := NewIntReq("RPUSHX", key, value) - c.Run(req) + c.Queue(req) return req } @@ -560,92 +560,92 @@ func (c *Client) RPushX(key string, value string) *IntReq { func (c *Client) SAdd(key string, members ...string) *IntReq { args := append([]string{"SADD", key}, members...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SCard(key string) *IntReq { req := NewIntReq("SCARD", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) SDiff(keys ...string) *MultiBulkReq { args := append([]string{"SDIFF"}, keys...) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SDiffStore(destination string, keys ...string) *IntReq { args := append([]string{"SDIFFSTORE", destination}, keys...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SInter(keys ...string) *MultiBulkReq { args := append([]string{"SINTER"}, keys...) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SInterStore(destination string, keys ...string) *IntReq { args := append([]string{"SINTERSTORE", destination}, keys...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SIsMember(key, member string) *BoolReq { req := NewBoolReq("SISMEMBER", key, member) - c.Run(req) + c.Queue(req) return req } func (c *Client) SMembers(key string) *MultiBulkReq { req := NewMultiBulkReq("SMEMBERS", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) SMove(source, destination, member string) *BoolReq { req := NewBoolReq("SMOVE", source, destination, member) - c.Run(req) + c.Queue(req) return req } func (c *Client) SPop(key string) *BulkReq { req := NewBulkReq("SPOP", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) SRandMember(key string) *BulkReq { req := NewBulkReq("SRANDMEMBER", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) SRem(key string, members ...string) *IntReq { args := append([]string{"SREM", key}, members...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SUnion(keys ...string) *MultiBulkReq { args := append([]string{"SUNION"}, keys...) req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) SUnionStore(destination string, keys ...string) *IntReq { args := append([]string{"SUNIONSTORE", destination}, keys...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -670,25 +670,25 @@ func (c *Client) ZAdd(key string, members ...*ZMember) *IntReq { args = append(args, m.ScoreString(), m.Member) } req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZCard(key string) *IntReq { req := NewIntReq("ZCARD", key) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZCount(key, min, max string) *IntReq { req := NewIntReq("ZCOUNT", key, min, max) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZIncrBy(key string, increment int64, member string) *IntReq { req := NewIntReq("ZINCRBY", key, strconv.FormatInt(increment, 10), member) - c.Run(req) + c.Queue(req) return req } @@ -711,7 +711,7 @@ func (c *Client) ZInterStore( args = append(args, "AGGREGATE", aggregate) } req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -726,7 +726,7 @@ func (c *Client) ZRange(key string, start, stop int64, withScores bool) *MultiBu args = append(args, "WITHSCORES") } req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -749,20 +749,20 @@ func (c *Client) ZRangeByScore( ) } req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZRank(key, member string) *IntNilReq { req := NewIntNilReq("ZRANK", key, member) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZRem(key string, members ...string) *IntReq { args := append([]string{"ZREM", key}, members...) req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -773,13 +773,13 @@ func (c *Client) ZRemRangeByRank(key string, start, stop int64) *IntReq { strconv.FormatInt(start, 10), strconv.FormatInt(stop, 10), ) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZRemRangeByScore(key, min, max string) *IntReq { req := NewIntReq("ZREMRANGEBYSCORE", key, min, max) - c.Run(req) + c.Queue(req) return req } @@ -789,7 +789,7 @@ func (c *Client) ZRevRange(key, start, stop string, withScores bool) *MultiBulkR args = append(args, "WITHSCORES") } req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -811,19 +811,19 @@ func (c *Client) ZRevRangeByScore( ) } req := NewMultiBulkReq(args...) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZRevRank(key, member string) *IntNilReq { req := NewIntNilReq("ZREVRANK", key, member) - c.Run(req) + c.Queue(req) return req } func (c *Client) ZScore(key, member string) *FloatReq { req := NewFloatReq("ZSCORE", key, member) - c.Run(req) + c.Queue(req) return req } @@ -846,7 +846,7 @@ func (c *Client) ZUnionStore( args = append(args, "AGGREGATE", aggregate) } req := NewIntReq(args...) - c.Run(req) + c.Queue(req) return req } @@ -858,7 +858,7 @@ func (c *Client) PubSubClient() *PubSubClient { func (c *Client) Publish(channel, message string) *IntReq { req := NewIntReq("PUBLISH", channel, message) - c.Run(req) + c.Queue(req) return req } diff --git a/redis.go b/redis.go index 58d775e..2237468 100644 --- a/redis.go +++ b/redis.go @@ -30,6 +30,8 @@ func NewClient(connect connectFunc, disconnect disconnectFunc) *Client { readerPool: bufreader.NewReaderPool(100, createReader), connect: connect, disconnect: disconnect, + + reqs: make([]Req, 0), } } @@ -99,14 +101,14 @@ func (c *Client) WriteRead(buf []byte, rd *bufreader.Reader) error { return c.ReadReply(rd) } -func (c *Client) Run(req Req) { - if c.reqs != nil { - c.mtx.Lock() - c.reqs = append(c.reqs, req) - c.mtx.Unlock() - return - } +func (c *Client) Queue(req Req) { + req.SetClient(c) + c.mtx.Lock() + c.reqs = append(c.reqs, req) + c.mtx.Unlock() +} +func (c *Client) Run(req Req) { rd, err := c.readerPool.Get() if err != nil { req.SetErr(err) @@ -128,23 +130,53 @@ func (c *Client) Run(req Req) { req.SetVal(val) } +func (c *Client) RunQueued() ([]Req, error) { + c.mtx.Lock() + reqs := c.reqs + c.reqs = make([]Req, 0) + c.mtx.Unlock() + + var multiReq []byte + if len(reqs) == 1 { + multiReq = reqs[0].Req() + } else { + multiReq = make([]byte, 0, 1024) + for _, req := range reqs { + multiReq = append(multiReq, req.Req()...) + } + } + + rd, err := c.readerPool.Get() + if err != nil { + return nil, err + } + defer c.readerPool.Add(rd) + + err = c.WriteRead(multiReq, rd) + if err != nil { + return nil, err + } + for _, req := range reqs { + val, err := req.ParseReply(rd) + if err != nil { + req.SetErr(err) + } else { + req.SetVal(val) + } + } + + return reqs, nil +} + //------------------------------------------------------------------------------ func (c *Client) Discard() { - if c.reqs == nil { - panic("MultiClient required") - } - c.mtx.Lock() c.reqs = c.reqs[:0] c.mtx.Unlock() } func (c *Client) Exec() ([]Req, error) { - if c.reqs == nil { - panic("MultiClient required") - } - c.mtx.Lock() reqs := c.reqs c.reqs = make([]Req, 0) @@ -170,20 +202,21 @@ func (c *Client) Exec() ([]Req, error) { statusReq := NewStatusReq() - // multi + // Parse MULTI command reply. _, err = statusReq.ParseReply(rd) if err != nil { return nil, err } + // Parse queued replies. for _ = range reqs { - // queue _, err = statusReq.ParseReply(rd) if err != nil { return nil, err } } + // Parse number of replies. line, err := rd.ReadLine('\n') if err != nil { return nil, err @@ -192,12 +225,14 @@ func (c *Client) Exec() ([]Req, error) { return nil, fmt.Errorf("Expected '*', but got line %q of %q.", line, rd.Bytes()) } + // Parse replies. for _, req := range reqs { val, err := req.ParseReply(rd) if err != nil { req.SetErr(err) + } else { + req.SetVal(val) } - req.SetVal(val) } return reqs, nil diff --git a/redis_test.go b/redis_test.go index 5f6cc1f..14db776 100644 --- a/redis_test.go +++ b/redis_test.go @@ -31,11 +31,11 @@ func (t *RedisTest) SetUpTest(c *C) { } t.redisC = redis.NewClient(connect, nil) - t.redisC.Flushdb() + t.redisC.Flushdb().Reply() } func (t *RedisTest) TearDownTest(c *C) { - t.redisC.Flushdb() + t.redisC.Flushdb().Reply() } //------------------------------------------------------------------------------ @@ -120,7 +120,7 @@ func (t *RedisTest) TestMove(c *C) { c.Check(err, IsNil) c.Check(isMoved, Equals, false) - t.redisC.Set("foo", "bar") + t.redisC.Set("foo", "bar").Reply() isMoved, err = t.redisC.Move("foo", 1).Reply() c.Check(err, IsNil) @@ -1353,6 +1353,27 @@ func (t *RedisTest) TestPubSub(c *C) { //------------------------------------------------------------------------------ +func (t *RedisTest) TestPipelining(c *C) { + t.redisC.Set("foo2", "bar2").Reply() + + setReq := t.redisC.Set("foo1", "bar1") + getReq := t.redisC.Get("foo2") + + reqs, err := t.redisC.RunQueued() + c.Check(err, IsNil) + c.Check(reqs, HasLen, 2) + + ok, err := setReq.Reply() + c.Check(err, IsNil) + c.Check(ok, Equals, "OK") + + v, err := getReq.Reply() + c.Check(err, IsNil) + c.Check(v, Equals, "bar2") +} + +//------------------------------------------------------------------------------ + func (t *RedisTest) TestDiscard(c *C) { multiC := t.redisC.Multi() diff --git a/request.go b/request.go index d76858c..6c99951 100644 --- a/request.go +++ b/request.go @@ -73,6 +73,7 @@ func PackReq(args []string) []byte { type Req interface { Req() []byte ParseReply(*bufreader.Reader) (interface{}, error) + SetClient(*Client) SetErr(error) SetVal(interface{}) } @@ -81,8 +82,13 @@ type Req interface { type BaseReq struct { args []string - val interface{} - err error + + client *Client + // TODO: use int32 and atomic access? + done bool + + val interface{} + err error } func NewBaseReq(args ...string) *BaseReq { @@ -95,10 +101,15 @@ func (r *BaseReq) Req() []byte { return PackReq(r.args) } +func (r *BaseReq) SetClient(c *Client) { + r.client = c +} + func (r *BaseReq) SetErr(err error) { if err == nil { panic("non-nil value expected") } + r.done = true r.err = err } @@ -106,6 +117,7 @@ func (r *BaseReq) SetVal(val interface{}) { if val == nil { panic("non-nil value expected") } + r.done = true r.val = val } @@ -141,6 +153,16 @@ func (r *StatusReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *StatusReq) Reply() (string, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return "", err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return "", r.err } @@ -175,6 +197,16 @@ func (r *IntReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *IntReq) Reply() (int64, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return 0, err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return 0, r.err } @@ -211,6 +243,16 @@ func (r *IntNilReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *IntNilReq) Reply() (int64, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return 0, err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return 0, r.err } @@ -245,6 +287,16 @@ func (r *BoolReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *BoolReq) Reply() (bool, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return false, err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return false, r.err } @@ -288,6 +340,16 @@ func (r *BulkReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *BulkReq) Reply() (string, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return "", err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return "", r.err } @@ -331,6 +393,16 @@ func (r *FloatReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *FloatReq) Reply() (float64, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return 0, err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return 0, r.err } @@ -416,6 +488,16 @@ func (r *MultiBulkReq) ParseReply(rd *bufreader.Reader) (interface{}, error) { } func (r *MultiBulkReq) Reply() ([]interface{}, error) { + if !r.done { + _, err := r.client.RunQueued() + if err != nil { + return nil, err + } + if !r.done { + panic("req is not ready") + } + } + if r.err != nil { return nil, r.err }