chore(txPipeline): refactor slottedCommands impl

osscluster.go

@@ -1526,46 +1526,26 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err
 		return err
 	}
 
-	cmdsMap := map[int][]Cmder{}
+	keyedCmdsBySlot := c.slottedKeyedCommands(cmds)
 	slot := -1
-	// get only the keyed commands
+	switch len(keyedCmdsBySlot) {
-	keyedCmds := c.keyedCmds(cmds)
+	case 0:
-	if len(keyedCmds) == 0 {
-		// no keyed commands try random slot
 		slot = hashtag.RandomSlot()
-	} else {
+	case 1:
-		// keyed commands, get slot from them
+		for sl := range keyedCmdsBySlot {
-		// if more than one slot, return cross slot error
-		cmdsBySlot := c.mapCmdsBySlot(keyedCmds)
-		if len(cmdsBySlot) > 1 {
-			// cross slot error, we have more than one slot for keyed commands
-			setCmdsErr(cmds, ErrCrossSlot)
-			return ErrCrossSlot
-		}
-		// get the slot, should be only one
-		for sl := range cmdsBySlot {
 			slot = sl
 			break
 		}
-	}
+	default:
-	// slot was not determined, try random one
-	if slot == -1 {
-		slot = hashtag.RandomSlot()
-	}
-	cmdsMap[slot] = cmds
-
 		// TxPipeline does not support cross slot transaction.
-	// double check the commands are in the same slot
-	if len(cmdsMap) > 1 {
 		setCmdsErr(cmds, ErrCrossSlot)
 		return ErrCrossSlot
 	}
 
-	for slot, cmds := range cmdsMap {
 	node, err := state.slotMasterNode(slot)
 	if err != nil {
 		setCmdsErr(cmds, err)
-			continue
+		return err
 	}
 
 	cmdsMap := map[*clusterNode][]Cmder{node: cmds}
@@ -1594,34 +1574,30 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err
 		}
 		cmdsMap = failedCmds.m
 	}
-	}
 
 	return cmdsFirstErr(cmds)
 }
 
-func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
+// slottedKeyedCommands returns a map of slot to commands taking into account
-	cmdsMap := make(map[int][]Cmder)
+// only commands that have keys.
-	preferredRandomSlot := -1
+func (c *ClusterClient) slottedKeyedCommands(cmds []Cmder) map[int][]Cmder {
-	for _, cmd := range cmds {
+	cmdsSlots := map[int][]Cmder{}
-		slot := c.cmdSlot(cmd, preferredRandomSlot)
-		if preferredRandomSlot == -1 {
-			preferredRandomSlot = slot
-		}
-		cmdsMap[slot] = append(cmdsMap[slot], cmd)
-	}
-	return cmdsMap
-}
 
-// keyedCmds returns all the keyed commands from the cmds slice
+	prefferedRandomSlot := -1
-// it determines keyed commands by checking if the command has a first key position
-func (c *ClusterClient) keyedCmds(cmds []Cmder) []Cmder {
-	keyedCmds := make([]Cmder, 0, len(cmds))
 	for _, cmd := range cmds {
-		if cmdFirstKeyPos(cmd) != 0 {
+		if cmdFirstKeyPos(cmd) == 0 {
-			keyedCmds = append(keyedCmds, cmd)
+			continue
 		}
+
+		slot := c.cmdSlot(cmd, prefferedRandomSlot)
+		if prefferedRandomSlot == -1 {
+			prefferedRandomSlot = slot
 		}
-	return keyedCmds
+
+		cmdsSlots[slot] = append(cmdsSlots[slot], cmd)
+	}
+
+	return cmdsSlots
 }
 
 func (c *ClusterClient) processTxPipelineNode(