Repository: go-redis/redis
Pull request #1103 (2019-07-25 10:28:54)

Propagate context in Pipeline


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cd9af8f..3728c9d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,12 +3,11 @@
 ## v7 WIP
 
 - WrapProcess is replaced with more convenient AddHook that has access to context.Context.
-- WithContext now can not be used to create a shallow copy of the client.
+- WithContext no longer creates shallow copy.
 - New methods ProcessContext, DoContext, and ExecContext.
 - Client respects Context.Deadline when setting net.Conn deadline.
-- Client listens on Context.Done while waiting for a connection from the pool and returns an error when context context is cancelled.
+- Client listens on Context.Done while waiting for a connection from the pool.
 - Add PubSub.ChannelWithSubscriptions that sends `*Subscription` in addition to `*Message` to allow detecting reconnections.
-- `time.Time` is now marshalled in RFC3339 format. `rdb.Get("foo").Time()` helper is added to parse time.
 
 ## v6.15
 
diff --git a/cluster.go b/cluster.go
index 26a668c..932d3d6 100644
--- a/cluster.go
+++ b/cluster.go
@@ -1023,6 +1023,7 @@ func (c *ClusterClient) reaper(idleCheckFrequency time.Duration) {
 
 func (c *ClusterClient) Pipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processPipeline,
 	}
 	pipe.init()
@@ -1220,6 +1221,7 @@ func (c *ClusterClient) checkMovedErr(
 // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
 func (c *ClusterClient) TxPipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processTxPipeline,
 	}
 	pipe.init()
diff --git a/command.go b/command.go
index 7b7c353..c302597 100644
--- a/command.go
+++ b/command.go
@@ -659,13 +659,6 @@ func (cmd *StringCmd) Float64() (float64, error) {
 	return strconv.ParseFloat(cmd.Val(), 64)
 }
 
-func (cmd *StringCmd) Time() (time.Time, error) {
-	if cmd.err != nil {
-		return time.Time{}, cmd.err
-	}
-	return time.Parse(time.RFC3339, cmd.Val())
-}
-
 func (cmd *StringCmd) Scan(val interface{}) error {
 	if cmd.err != nil {
 		return cmd.err
diff --git a/commands.go b/commands.go
index 710aef2..26af559 100644
--- a/commands.go
+++ b/commands.go
@@ -42,9 +42,7 @@ func appendArgs(dst, src []interface{}) []interface{} {
 		}
 	}
 
-	for _, v := range src {
-		dst = append(dst, v)
-	}
+	dst = append(dst, src...)
 	return dst
 }
 
diff --git a/internal/proto/writer.go b/internal/proto/writer.go
index cded347..d106ce0 100644
--- a/internal/proto/writer.go
+++ b/internal/proto/writer.go
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io"
 	"strconv"
-	"time"
 
 	"github.com/go-redis/redis/internal/util"
 )
@@ -93,8 +92,6 @@ func (w *Writer) writeArg(v interface{}) error {
 		} else {
 			return w.int(0)
 		}
-	case time.Time:
-		return w.string(v.Format(time.RFC3339))
 	case encoding.BinaryMarshaler:
 		b, err := v.MarshalBinary()
 		if err != nil {
diff --git a/pipeline.go b/pipeline.go
index 8b9147e..3c861cd 100644
--- a/pipeline.go
+++ b/pipeline.go
@@ -41,6 +41,7 @@ type Pipeline struct {
 	cmdable
 	statefulCmdable
 
+	ctx  context.Context
 	exec pipelineExecer
 
 	mu     sync.Mutex
@@ -98,7 +99,7 @@ func (c *Pipeline) discard() error {
 // Exec always returns list of commands and error of the first failed
 // command if any.
 func (c *Pipeline) Exec() ([]Cmder, error) {
-	return c.ExecContext(context.Background())
+	return c.ExecContext(c.ctx)
 }
 
 func (c *Pipeline) ExecContext(ctx context.Context) ([]Cmder, error) {
diff --git a/redis.go b/redis.go
index 10200bb..abdda7e 100644
--- a/redis.go
+++ b/redis.go
@@ -136,7 +136,7 @@ func (c *baseClient) newConn(ctx context.Context) (*pool.Conn, error) {
 		return nil, err
 	}
 
-	err = c.initConn(cn)
+	err = c.initConn(ctx, cn)
 	if err != nil {
 		_ = c.connPool.CloseConn(cn)
 		return nil, err
@@ -169,7 +169,7 @@ func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
 		return nil, err
 	}
 
-	err = c.initConn(cn)
+	err = c.initConn(ctx, cn)
 	if err != nil {
 		c.connPool.Remove(cn)
 		return nil, err
@@ -202,7 +202,7 @@ func (c *baseClient) releaseConnStrict(cn *pool.Conn, err error) {
 	}
 }
 
-func (c *baseClient) initConn(cn *pool.Conn) error {
+func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 	if cn.Inited {
 		return nil
 	}
@@ -215,7 +215,7 @@ func (c *baseClient) initConn(cn *pool.Conn) error {
 		return nil
 	}
 
-	conn := newConn(c.opt, cn)
+	conn := newConn(ctx, c.opt, cn)
 	_, err := conn.Pipelined(func(pipe Pipeliner) error {
 		if c.opt.Password != "" {
 			pipe.Auth(c.opt.Password)
@@ -547,6 +547,7 @@ func (c *Client) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
 
 func (c *Client) Pipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processPipeline,
 	}
 	pipe.init()
@@ -560,6 +561,7 @@ func (c *Client) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
 // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
 func (c *Client) TxPipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processTxPipeline,
 	}
 	pipe.init()
@@ -625,19 +627,27 @@ func (c *Client) PSubscribe(channels ...string) *PubSub {
 
 //------------------------------------------------------------------------------
 
-// Conn is like Client, but its pool contains single connection.
-type Conn struct {
+type conn struct {
 	baseClient
 	cmdable
 	statefulCmdable
 }
 
-func newConn(opt *Options, cn *pool.Conn) *Conn {
+// Conn is like Client, but its pool contains single connection.
+type Conn struct {
+	*conn
+	ctx context.Context
+}
+
+func newConn(ctx context.Context, opt *Options, cn *pool.Conn) *Conn {
 	c := Conn{
-		baseClient: baseClient{
-			opt:      opt,
-			connPool: pool.NewSingleConnPool(cn),
+		conn: &conn{
+			baseClient: baseClient{
+				opt:      opt,
+				connPool: pool.NewSingleConnPool(cn),
+			},
 		},
+		ctx: ctx,
 	}
 	c.cmdable = c.Process
 	c.statefulCmdable = c.Process
@@ -645,7 +655,7 @@ func newConn(opt *Options, cn *pool.Conn) *Conn {
 }
 
 func (c *Conn) Process(cmd Cmder) error {
-	return c.ProcessContext(context.TODO(), cmd)
+	return c.ProcessContext(c.ctx, cmd)
 }
 
 func (c *Conn) ProcessContext(ctx context.Context, cmd Cmder) error {
@@ -658,6 +668,7 @@ func (c *Conn) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
 
 func (c *Conn) Pipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processPipeline,
 	}
 	pipe.init()
@@ -671,6 +682,7 @@ func (c *Conn) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
 // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
 func (c *Conn) TxPipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processTxPipeline,
 	}
 	pipe.init()
diff --git a/ring.go b/ring.go
index f8b5e5a..df71cc8 100644
--- a/ring.go
+++ b/ring.go
@@ -579,6 +579,7 @@ func (c *Ring) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
 
 func (c *Ring) Pipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processPipeline,
 	}
 	pipe.init()
@@ -597,6 +598,7 @@ func (c *Ring) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
 
 func (c *Ring) TxPipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processTxPipeline,
 	}
 	pipe.init()
diff --git a/tx.go b/tx.go
index bcaebde..38a8855 100644
--- a/tx.go
+++ b/tx.go
@@ -93,7 +93,7 @@ func (c *Tx) Watch(keys ...string) *StatusCmd {
 		args[1+i] = key
 	}
 	cmd := NewStatusCmd(args...)
-	c.Process(cmd)
+	_ = c.Process(cmd)
 	return cmd
 }
 
@@ -105,13 +105,14 @@ func (c *Tx) Unwatch(keys ...string) *StatusCmd {
 		args[1+i] = key
 	}
 	cmd := NewStatusCmd(args...)
-	c.Process(cmd)
+	_ = c.Process(cmd)
 	return cmd
 }
 
 // Pipeline creates a new pipeline. It is more convenient to use Pipelined.
 func (c *Tx) Pipeline() Pipeliner {
 	pipe := Pipeline{
+		ctx:  c.ctx,
 		exec: c.processTxPipeline,
 	}
 	pipe.init()


Test patch:
diff --git a/command_test.go b/command_test.go
index 54e9945..020f8aa 100644
--- a/command_test.go
+++ b/command_test.go
@@ -1,8 +1,6 @@
 package redis_test
 
 import (
-	"time"
-
 	"github.com/go-redis/redis"
 
 	. "github.com/onsi/ginkgo"
@@ -69,19 +67,4 @@ var _ = Describe("Cmd", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(val).To(Equal(f))
 	})
-
-	It("supports time.Time", func() {
-		tm := time.Date(2019, 01, 01, 0, 0, 0, 0, time.UTC)
-
-		err := client.Set("time_key", tm, 0).Err()
-		Expect(err).NotTo(HaveOccurred())
-
-		s, err := client.Get("time_key").Result()
-		Expect(err).NotTo(HaveOccurred())
-		Expect(s).To(Equal("2019-01-01T00:00:00Z"))
-
-		tm2, err := client.Get("time_key").Time()
-		Expect(err).NotTo(HaveOccurred())
-		Expect(tm2).To(BeTemporally("==", tm))
-	})
 })
diff --git a/internal/proto/write_buffer_test.go b/internal/proto/write_buffer_test.go
index 2de532e..29aa469 100644
--- a/internal/proto/write_buffer_test.go
+++ b/internal/proto/write_buffer_test.go
@@ -2,7 +2,6 @@ package proto_test
 
 import (
 	"bytes"
-	"encoding"
 	"io/ioutil"
 	"testing"
 	"time"
@@ -13,14 +12,6 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-type MyType struct{}
-
-var _ encoding.BinaryMarshaler = (*MyType)(nil)
-
-func (t *MyType) MarshalBinary() ([]byte, error) {
-	return []byte("hello"), nil
-}
-
 var _ = Describe("WriteBuffer", func() {
 	var buf *bytes.Buffer
 	var wr *proto.Writer
@@ -54,25 +45,16 @@ var _ = Describe("WriteBuffer", func() {
 			"\r\n")))
 	})
 
-	It("should append time", func() {
-		err := wr.WriteArgs([]interface{}{time.Unix(1414141414, 0).UTC()})
-		Expect(err).NotTo(HaveOccurred())
-
-		err = wr.Flush()
-		Expect(err).NotTo(HaveOccurred())
-
-		Expect(buf.Len()).To(Equal(31))
-	})
-
 	It("should append marshalable args", func() {
-		err := wr.WriteArgs([]interface{}{&MyType{}})
+		err := wr.WriteArgs([]interface{}{time.Unix(1414141414, 0)})
 		Expect(err).NotTo(HaveOccurred())
 
 		err = wr.Flush()
 		Expect(err).NotTo(HaveOccurred())
 
-		Expect(buf.Len()).To(Equal(15))
+		Expect(buf.Len()).To(Equal(26))
 	})
+
 })
 
 func BenchmarkWriteBuffer_Append(b *testing.B) {