Reuse single Pipeline type in Client, ClusterClient and Ring.

cluster.go

@@ -316,6 +316,92 @@ func (c *ClusterClient) reaper(frequency time.Duration) {
 	}
 }
 
+func (c *ClusterClient) Pipeline() *Pipeline {
+	pipe := &Pipeline{
+		exec: c.pipelineExec,
+	}
+	pipe.commandable.process = pipe.process
+	return pipe
+}
+
+func (c *ClusterClient) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
+	return c.Pipeline().pipelined(fn)
+}
+
+func (c *ClusterClient) pipelineExec(cmds []Cmder) error {
+	var retErr error
+
+	cmdsMap := make(map[string][]Cmder)
+	for _, cmd := range cmds {
+		slot := hashtag.Slot(cmd.clusterKey())
+		addr := c.slotMasterAddr(slot)
+		cmdsMap[addr] = append(cmdsMap[addr], cmd)
+	}
+
+	for attempt := 0; attempt <= c.opt.getMaxRedirects(); attempt++ {
+		failedCmds := make(map[string][]Cmder)
+
+		for addr, cmds := range cmdsMap {
+			client, err := c.getClient(addr)
+			if err != nil {
+				setCmdsErr(cmds, err)
+				retErr = err
+				continue
+			}
+
+			cn, err := client.conn()
+			if err != nil {
+				setCmdsErr(cmds, err)
+				retErr = err
+				continue
+			}
+
+			failedCmds, err = c.execClusterCmds(cn, cmds, failedCmds)
+			if err != nil {
+				retErr = err
+			}
+			client.putConn(cn, err, false)
+		}
+
+		cmdsMap = failedCmds
+	}
+
+	return retErr
+}
+
+func (c *ClusterClient) execClusterCmds(
+	cn *pool.Conn, cmds []Cmder, failedCmds map[string][]Cmder,
+) (map[string][]Cmder, error) {
+	if err := writeCmd(cn, cmds...); err != nil {
+		setCmdsErr(cmds, err)
+		return failedCmds, err
+	}
+
+	var firstCmdErr error
+	for i, cmd := range cmds {
+		err := cmd.readReply(cn)
+		if err == nil {
+			continue
+		}
+		if isNetworkError(err) {
+			cmd.reset()
+			failedCmds[""] = append(failedCmds[""], cmds[i:]...)
+			break
+		} else if moved, ask, addr := isMovedError(err); moved {
+			c.lazyReloadSlots()
+			cmd.reset()
+			failedCmds[addr] = append(failedCmds[addr], cmd)
+		} else if ask {
+			cmd.reset()
+			failedCmds[addr] = append(failedCmds[addr], NewCmd("ASKING"), cmd)
+		} else if firstCmdErr == nil {
+			firstCmdErr = err
+		}
+	}
+
+	return failedCmds, firstCmdErr
+}
+
 //------------------------------------------------------------------------------
 
 // ClusterOptions are used to configure a cluster client and should be

cluster_test.go

@@ -449,7 +449,7 @@ var _ = Describe("Cluster", func() {
 			Expect(client.Set("C", "C_value", 0).Err()).NotTo(HaveOccurred())
 
 			var a, b, c *redis.StringCmd
-			cmds, err := client.Pipelined(func(pipe *redis.ClusterPipeline) error {
+			cmds, err := client.Pipelined(func(pipe *redis.Pipeline) error {
 				a = pipe.Get("A")
 				b = pipe.Get("B")
 				c = pipe.Get("C")

pipeline.go

@@ -13,33 +13,14 @@ import (
 type Pipeline struct {
 	commandable
 
-	client baseClient
+	exec func([]Cmder) error
 
 	mu   sync.Mutex // protects cmds
 	cmds []Cmder
 
 	closed int32
 }
 
-func (c *Client) Pipeline() *Pipeline {
-	pipe := &Pipeline{
-		client: c.baseClient,
-		cmds:   make([]Cmder, 0, 10),
-	}
-	pipe.commandable.process = pipe.process
-	return pipe
-}
-
-func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
-	pipe := c.Pipeline()
-	if err := fn(pipe); err != nil {
-		return nil, err
-	}
-	cmds, err := pipe.Exec()
-	_ = pipe.Close()
-	return cmds, err
-}
-
 func (pipe *Pipeline) process(cmd Cmder) {
 	pipe.mu.Lock()
 	pipe.cmds = append(pipe.cmds, cmd)
@@ -73,7 +54,7 @@ func (pipe *Pipeline) Discard() error {
 //
 // Exec always returns list of commands and error of the first failed
 // command if any.
-func (pipe *Pipeline) Exec() (cmds []Cmder, retErr error) {
+func (pipe *Pipeline) Exec() ([]Cmder, error) {
 	if pipe.isClosed() {
 		return nil, pool.ErrClosed
 	}
@@ -85,31 +66,19 @@ func (pipe *Pipeline) Exec() (cmds []Cmder, retErr error) {
 		return pipe.cmds, nil
 	}
 
-	cmds = pipe.cmds
-	pipe.cmds = make([]Cmder, 0, 10)
+	cmds := pipe.cmds
+	pipe.cmds = nil
 
-	failedCmds := cmds
-	for i := 0; i <= pipe.client.opt.MaxRetries; i++ {
-		cn, err := pipe.client.conn()
-		if err != nil {
-			setCmdsErr(failedCmds, err)
-			return cmds, err
-		}
+	return cmds, pipe.exec(cmds)
+}
 
-		if i > 0 {
-			resetCmds(failedCmds)
-		}
-		failedCmds, err = execCmds(cn, failedCmds)
-		pipe.client.putConn(cn, err, false)
-		if err != nil && retErr == nil {
-			retErr = err
-		}
-		if len(failedCmds) == 0 {
-			break
-		}
+func (pipe *Pipeline) pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
+	if err := fn(pipe); err != nil {
+		return nil, err
 	}
-
-	return cmds, retErr
+	cmds, err := pipe.Exec()
+	_ = pipe.Close()
+	return cmds, err
 }
 
 func execCmds(cn *pool.Conn, cmds []Cmder) ([]Cmder, error) {

redis.go

@@ -174,3 +174,40 @@ func (c *Client) PoolStats() *PoolStats {
 		FreeConns:  s.FreeConns,
 	}
 }
+
+func (c *Client) Pipeline() *Pipeline {
+	pipe := &Pipeline{
+		exec: c.pipelineExec,
+	}
+	pipe.commandable.process = pipe.process
+	return pipe
+}
+
+func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
+	return c.Pipeline().pipelined(fn)
+}
+
+func (c *Client) pipelineExec(cmds []Cmder) error {
+	var retErr error
+	failedCmds := cmds
+	for i := 0; i <= c.opt.MaxRetries; i++ {
+		cn, err := c.conn()
+		if err != nil {
+			setCmdsErr(failedCmds, err)
+			return err
+		}
+
+		if i > 0 {
+			resetCmds(failedCmds)
+		}
+		failedCmds, err = execCmds(cn, failedCmds)
+		c.putConn(cn, err, false)
+		if err != nil && retErr == nil {
+			retErr = err
+		}
+		if len(failedCmds) == 0 {
+			break
+		}
+	}
+	return retErr
+}