pipeline: Add DiscardQueued method.

pipeline.go

@@ -30,6 +30,12 @@ func (c *PipelineClient) Close() error {
 	return nil
 }
 
+func (c *PipelineClient) DiscardQueued() {
+	c.mtx.Lock()
+	c.reqs = c.reqs[:0]
+	c.mtx.Unlock()
+}
+
 func (c *PipelineClient) RunQueued() ([]Req, error) {
 	c.mtx.Lock()
 	if len(c.reqs) == 0 {

redis_test.go

@@ -2184,6 +2184,20 @@ func (t *RedisTest) TestPipeline(c *C) {
 	c.Assert(getNil.Val(), Equals, "")
 }
 
+func (t *RedisTest) TestPipelineDiscardQueued(c *C) {
+	pipeline, err := t.client.PipelineClient()
+	c.Assert(err, IsNil)
+	defer func() {
+		c.Assert(pipeline.Close(), IsNil)
+	}()
+
+	pipeline.Get("key")
+	pipeline.DiscardQueued()
+	reqs, err := pipeline.RunQueued()
+	c.Assert(err, IsNil)
+	c.Assert(reqs, HasLen, 0)
+}
+
 func (t *RedisTest) TestPipelineFunc(c *C) {
 	var get *redis.StringReq
 	reqs, err := t.client.Pipelined(func(c *redis.PipelineClient) {