chore(err): Extract crossslot err and add test

Author: ndyakov

error.go

@@ -22,6 +22,12 @@ var ErrPoolExhausted = pool.ErrPoolExhausted
 // ErrPoolTimeout timed out waiting to get a connection from the connection pool.
 var ErrPoolTimeout = pool.ErrPoolTimeout
 
+// ErrCrossSlot is returned when keys are used in the same Redis command and
+// the keys are not in the same hash slot. This error is returned by Redis
+// Cluster and will be returned by the client when TxPipeline or TxPipelined
+// is used on a ClusterClient with keys in different slots.
+var ErrCrossSlot = proto.RedisError("CROSSSLOT Keys in request don't hash to the same slot")
+
 // HasErrorPrefix checks if the err is a Redis error and the message contains a prefix.
 func HasErrorPrefix(err error, prefix string) bool {
 	var rErr Error

osscluster.go

@@ -1506,9 +1506,8 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err
 	cmdsMap := c.mapCmdsBySlot(cmds)
 	// TxPipeline does not support cross slot transaction.
 	if len(cmdsMap) > 1 {
-		err := fmt.Errorf("redis: CROSSSLOT Keys in request don't hash to the same slot")
-		setCmdsErr(cmds, err)
-		return err
+		setCmdsErr(cmds, ErrCrossSlot)
+		return ErrCrossSlot
 	}
 	if len(cmdsMap) == 0 {
 		return nil

osscluster_test.go

@@ -593,6 +593,20 @@ var _ = Describe("ClusterClient", func() {
 				// Use hashtag to force all keys to the same slot.
 				keys := []string{"A{s}", "B{s}", "C{s}", "D{s}", "E{s}", "F{s}", "G{s}"}
 				assertPipeline(keys)
+
+				// make sure CrossSlot error is returned
+				It("returns CrossSlot error", func() {
+					pipe.Set(ctx, "A{s}", "A_value", 0)
+					pipe.Set(ctx, "B{t}", "B_value", 0)
+					_, err := pipe.Exec(ctx)
+					Expect(err).To(MatchError(redis.ErrCrossSlot))
+				})
+
+				// doesn't fail when no commands are queued
+				It("returns no error when there are no commands", func() {
+					_, err := pipe.Exec(ctx)
+					Expect(err).NotTo(HaveOccurred())
+				})
 			})
 		})