Support redis.RESP3 protocol

cluster.go

@@ -1178,8 +1178,8 @@ func (c *ClusterClient) _processPipelineNode(
 				return err
 			}
 
-			return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
-				return c.pipelineReadCmds(ctx, node, rd, cmds, failedCmds)
+			return cn.WithReaders(ctx, c.opt.ReadTimeout, len(cmds), func(pvs []*proto.Value) error {
+				return c.pipelineReadCmds(ctx, node, pvs, cmds, failedCmds)
 			})
 		})
 	})
@@ -1188,12 +1188,15 @@ func (c *ClusterClient) _processPipelineNode(
 func (c *ClusterClient) pipelineReadCmds(
 	ctx context.Context,
 	node *clusterNode,
-	rd *proto.Reader,
+	pvs []*proto.Value,
 	cmds []Cmder,
 	failedCmds *cmdsMap,
 ) error {
-	for _, cmd := range cmds {
-		err := cmd.readReply(rd)
+	if len(pvs) != len(cmds) {
+		return fmt.Errorf("value len(%d), cmds len(%d)", len(pvs), len(cmds))
+	}
+	for key, cmd := range cmds {
+		err := cmd.readReply(pvs[key])
 		cmd.SetErr(err)
 
 		if err == nil {
@@ -1343,12 +1346,12 @@ func (c *ClusterClient) _processTxPipelineNode(
 				return err
 			}
 
-			return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
+			return cn.WithReaders(ctx, c.opt.ReadTimeout, len(cmds), func(pvs []*proto.Value) error {
 				statusCmd := cmds[0].(*StatusCmd)
 				// Trim multi and exec.
 				cmds = cmds[1 : len(cmds)-1]
 
-				err := c.txPipelineReadQueued(ctx, rd, statusCmd, cmds, failedCmds)
+				err := c.txPipelineReadQueued(ctx, pvs, statusCmd, cmds, failedCmds)
 				if err != nil {
 					moved, ask, addr := isMovedError(err)
 					if moved || ask {
@@ -1357,48 +1360,41 @@ func (c *ClusterClient) _processTxPipelineNode(
 					return err
 				}
 
-				return pipelineReadCmds(rd, cmds)
+				return pipelineReadCmds(pvs[len(pvs)-1].Slice, cmds)
 			})
 		})
 	})
 }
 
 func (c *ClusterClient) txPipelineReadQueued(
 	ctx context.Context,
-	rd *proto.Reader,
+	pvs []*proto.Value,
 	statusCmd *StatusCmd,
 	cmds []Cmder,
 	failedCmds *cmdsMap,
 ) error {
 	// Parse queued replies.
-	if err := statusCmd.readReply(rd); err != nil {
+	if err := statusCmd.readReply(pvs[0]); err != nil {
 		return err
 	}
 
-	for _, cmd := range cmds {
-		err := statusCmd.readReply(rd)
-		if err == nil || c.checkMovedErr(ctx, cmd, err, failedCmds) || isRedisError(err) {
-			continue
+	for key, cmd := range cmds {
+		err := statusCmd.readReply(pvs[key+1])
+		if err != nil && !c.checkMovedErr(ctx, cmd, err, failedCmds) && !isRedisError(err) {
+			return err
 		}
-		return err
 	}
 
-	// Parse number of replies.
-	line, err := rd.ReadLine()
-	if err != nil {
-		if err == Nil {
-			err = TxFailedErr
-		}
+	n, err := pvs[len(pvs)-1].SliceLen()
+	switch err {
+	case nil:
+	case Nil:
+		return TxFailedErr
+	default:
 		return err
 	}
-
-	switch line[0] {
-	case proto.ErrorReply:
-		return proto.ParseErrorReply(line)
-	case proto.ArrayReply:
-		// ok
-	default:
-		return fmt.Errorf("redis: expected '*', but got line %q", line)
+	if n != len(cmds) {
+		return fmt.Errorf("redis: expected %d, but got %d", len(cmds), n)
 	}
 
 	return nil