Merge pull request mediocregopher#180 from maciej/drain-pipeline

Drain pipeline on decode error

Author: Brian Picciano

action.go

@@ -441,14 +441,23 @@ func (p pipeline) Run(c Conn) error {
 	if err := c.Encode(p); err != nil {
 		return err
 	}
-	for _, cmd := range p {
+
+	for i, cmd := range p {
 		if err := c.Decode(cmd); err != nil {
+			p.drain(c, len(p)-i-1)
 			return decodeErr(cmd, err)
 		}
 	}
 	return nil
 }
 
+func (p pipeline) drain(c Conn, n int) {
+	rcv := resp2.Any{I: nil}
+	for i := 0; i < n; i++ {
+		_ = c.Decode(&rcv)
+	}
+}
+
 func decodeErr(cmd CmdAction, err error) error {
 	c, ok := cmd.(*cmdAction)
 	if ok {

action_test.go

@@ -6,9 +6,10 @@ import (
 	"fmt"
 	. "testing"
 
-	"github.com/mediocregopher/radix/v3/resp/resp2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/mediocregopher/radix/v3/resp/resp2"
 )
 
 func TestCmdAction(t *T) {
@@ -224,6 +225,37 @@ func TestPipelineAction(t *T) {
 			assert.Equal(t, ss[i], out[i])
 		}
 	}
+
+	t.Run("drain on decodeErr", func(t *T) {
+		// Setup
+		k1 := randStr()
+		k2 := randStr()
+		kvs := map[string]string{
+			k1: randStr(),
+			k2: randStr(),
+		}
+
+		for k, v := range kvs {
+			require.NoError(t, c.Do(Cmd(nil, "SET", k, v)))
+		}
+
+		var intRcv int
+		var strRcv string
+
+		pipeline := Pipeline(
+			Cmd(&intRcv, "GET", k1),
+			Cmd(nil, "GET", k2),
+		)
+
+		err := c.Do(pipeline)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "failed to decode")
+
+		err = c.Do(Cmd(&strRcv, "GET", k1))
+		require.NoError(t, err)
+
+		assert.Equal(t, kvs[k1], strRcv)
+	})
 }
 
 func ExamplePipeline() {