Merge branch 'master' into ndyakov/CAE-1088-resp3-notification-handlers

Author: ndyakov

.github/release-drafter-config.yml

@@ -36,6 +36,8 @@ categories:
 change-template: '- $TITLE (#$NUMBER)'
 exclude-labels:
   - 'skip-changelog'
+exclude-contributors:
+  - 'dependabot'
 template: |
   # Changes
 

osscluster.go

@@ -1749,10 +1749,16 @@ func (c *ClusterClient) txPipelineReadQueued(
 			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
 		}
 		err := statusCmd.readReply(rd)
-		if err == nil || c.checkMovedErr(ctx, cmd, err, failedCmds) || isRedisError(err) {
-			continue
+		if err != nil {
+			if c.checkMovedErr(ctx, cmd, err, failedCmds) {
+				// will be processed later
+				continue
+			}
+			cmd.SetErr(err)
+			if !isRedisError(err) {
+				return err
+			}
 		}
-		return err
 	}
 
 	// To be sure there are no buffered push notifications, we process them before reading the reply

pipeline.go

@@ -30,9 +30,12 @@ type Pipeliner interface {
 	// If a certain Redis command is not yet supported, you can use Do to execute it.
 	Do(ctx context.Context, args ...interface{}) *Cmd
 
-	// Process puts the commands to be executed into the pipeline buffer.
+	// Process queues the cmd for later execution.
 	Process(ctx context.Context, cmd Cmder) error
 
+	// BatchProcess adds multiple commands to be executed into the pipeline buffer.
+	BatchProcess(ctx context.Context, cmd ...Cmder) error
+
 	// Discard discards all commands in the pipeline buffer that have not yet been executed.
 	Discard()
 
@@ -79,7 +82,12 @@ func (c *Pipeline) Do(ctx context.Context, args ...interface{}) *Cmd {
 
 // Process queues the cmd for later execution.
 func (c *Pipeline) Process(ctx context.Context, cmd Cmder) error {
-	c.cmds = append(c.cmds, cmd)
+	return c.BatchProcess(ctx, cmd)
+}
+
+// BatchProcess queues multiple cmds for later execution.
+func (c *Pipeline) BatchProcess(ctx context.Context, cmd ...Cmder) error {
+	c.cmds = append(c.cmds, cmd...)
 	return nil
 }
 

pipeline_test.go

@@ -114,6 +114,25 @@ var _ = Describe("pipelining", func() {
 			err := pipe.Do(ctx).Err()
 			Expect(err).To(Equal(errors.New("redis: please enter the command to be executed")))
 		})
+
+		It("should process", func() {
+			err := pipe.Process(ctx, redis.NewCmd(ctx, "asking"))
+			Expect(err).To(BeNil())
+			Expect(pipe.Cmds()).To(HaveLen(1))
+		})
+
+		It("should batchProcess", func() {
+			err := pipe.BatchProcess(ctx, redis.NewCmd(ctx, "asking"))
+			Expect(err).To(BeNil())
+			Expect(pipe.Cmds()).To(HaveLen(1))
+
+			pipe.Discard()
+			Expect(pipe.Cmds()).To(HaveLen(0))
+
+			err = pipe.BatchProcess(ctx, redis.NewCmd(ctx, "asking"), redis.NewCmd(ctx, "set", "key", "value"))
+			Expect(err).To(BeNil())
+			Expect(pipe.Cmds()).To(HaveLen(2))
+		})
 	}
 
 	Describe("Pipeline", func() {

redis.go

@@ -764,6 +764,7 @@ func (c *baseClient) generalProcessPipeline(
 			return err
 		})
 		if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) {
+			setCmdsErr(cmds, lastErr)
 			return lastErr
 		}
 	}
@@ -859,13 +860,16 @@ func (c *baseClient) txPipelineReadQueued(ctx context.Context, cn *pool.Conn, rd
 	}
 
 	// Parse +QUEUED.
-	for range cmds {
+  for _, cmd := range cmds {
 		// To be sure there are no buffered push notifications, we process them before reading the reply
 		if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
 			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
 		}
-		if err := statusCmd.readReply(rd); err != nil && !isRedisError(err) {
-			return err
+		if err := statusCmd.readReply(rd); err != nil {
+			cmd.SetErr(err)
+			if !isRedisError(err) {
+				return err
+			}
 		}
 	}