Merge pull request #2294 from go-redis/fix/conn-tx-pipeline

fix: wrap cmds in Conn.TxPipeline

redis.go

@@ -725,21 +725,13 @@ func (c *Conn) Process(ctx context.Context, cmd Cmder) error {
 	return err
 }
 
-func (c *Conn) processPipeline(ctx context.Context, cmds []Cmder) error {
-	return c.hooks.processPipeline(ctx, cmds)
-}
-
-func (c *Conn) processTxPipeline(ctx context.Context, cmds []Cmder) error {
-	return c.hooks.processTxPipeline(ctx, cmds)
-}
-
 func (c *Conn) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
 	return c.Pipeline().Pipelined(ctx, fn)
 }
 
 func (c *Conn) Pipeline() Pipeliner {
 	pipe := Pipeline{
-		exec: c.processPipeline,
+		exec: c.hooks.processPipeline,
 	}
 	pipe.init()
 	return &pipe
@@ -752,7 +744,10 @@ func (c *Conn) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmd
 // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
 func (c *Conn) TxPipeline() Pipeliner {
 	pipe := Pipeline{
-		exec: c.processTxPipeline,
+		exec: func(ctx context.Context, cmds []Cmder) error {
+			cmds = wrapMultiExec(ctx, cmds)
+			return c.hooks.processTxPipeline(ctx, cmds)
+		},
 	}
 	pipe.init()
 	return &pipe

redis_test.go

@@ -446,3 +446,25 @@ var _ = Describe("Client context cancelation", func() {
 		Expect(err).To(BeIdenticalTo(context.Canceled))
 	})
 })
+
+var _ = Describe("Conn", func() {
+	var client *redis.Client
+
+	BeforeEach(func() {
+		client = redis.NewClient(redisOptions())
+		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		err := client.Close()
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("TxPipeline", func() {
+		tx := client.Conn().TxPipeline()
+		tx.SwapDB(ctx, 0, 2)
+		tx.SwapDB(ctx, 1, 0)
+		_, err := tx.Exec(ctx)
+		Expect(err).NotTo(HaveOccurred())
+	})
+})