From 5053db2f9c8b3ca25f497a75f70012c7ad6cd775 Mon Sep 17 00:00:00 2001
From: Vladimir Mihailenco <vladimir.webdev@gmail.com>
Date: Tue, 22 Nov 2022 14:28:39 +0200
Subject: [PATCH] fix: wrap cmds in Conn.TxPipeline

---
 redis.go      | 15 +++++----------
 redis_test.go | 22 ++++++++++++++++++++++
 2 files changed, 27 insertions(+), 10 deletions(-)

diff --git a/redis.go b/redis.go
index 0494091..ee6c9bd 100644
--- a/redis.go
+++ b/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
diff --git a/redis_test.go b/redis_test.go
index d0ccb45..8df1dd8 100644
--- a/redis_test.go
+++ b/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())
+	})
+})