Merge pull request #35 from exuan/pipeline

chore: TairCmdable add TairPipeline

.github/workflows/go.yml

@@ -79,4 +79,5 @@ jobs:
           go test -v ./... -run TestTairStringTestSuite/TestTairStringTestSuite
           go test -v ./... -run TestTairStringTestSuite/TestTairHashTestSuite
           go test -v ./... -run TestTairStringTestSuite/TestTairZsetTestSuite
+          go test -v ./... -run TestTairPipelineTestSuite
 

tair/taircluster.go

@@ -30,3 +30,15 @@ func NewTairClusterClient(opt *TairClusterOptions) *TairClusterClient {
 	tc.tairCmdable = tc.Process
 	return tc
 }
+
+func (t *TairClusterClient) TairPipeline() TairPipeline {
+	pipe := TairPipeline{
+		Pipeline: t.ClusterClient.Pipeline().(*redis.Pipeline),
+	}
+	pipe.init()
+	return pipe
+}
+
+func (t *TairClusterClient) TairPipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error) {
+	return t.ClusterClient.Pipeline().Pipelined(ctx, fn)
+}

tair/taircommands.go

@@ -163,6 +163,9 @@ type TairCmdable interface {
 	BfMExists(ctx context.Context, key string, items ...string) *redis.BoolSliceCmd
 	BfInsert(ctx context.Context, key string, bfInsertArgs *BfInsertArgs, items ...string) *redis.BoolSliceCmd
 	BfDebug(ctx context.Context, key string) *redis.StringSliceCmd
+	// TairPipeline
+	TairPipeline() TairPipeline
+	TairPipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error)
 }
 
 func toMs(dur time.Duration) int64 {

tair/tairpipeline_test.go

@@ -0,0 +1,70 @@
+package tair_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/alibaba/tair-go/tair"
+	"github.com/go-redis/redis/v8"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
+)
+
+type PipelineTestSuite struct {
+	suite.Suite
+	tairClient        *tair.TairClient
+	tairClusterClient *tair.TairClusterClient
+}
+
+func (suite *PipelineTestSuite) SetupTest() {
+	suite.tairClient = tair.NewTairClient(redisOptions())
+	assert.Equal(suite.T(), "OK", suite.tairClient.FlushDB(ctx).Val())
+
+	suite.tairClusterClient = cluster.newClusterClient(ctx, redisClusterOptions())
+	err := suite.tairClusterClient.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
+		return master.FlushDB(ctx).Err()
+	})
+	assert.NoError(suite.T(), err)
+}
+
+func (suite *PipelineTestSuite) TestTairPipeline() {
+	pipe := suite.tairClient.TairPipeline()
+	pipe.Set(ctx, "key", "value", 0)
+	pipe.Get(ctx, "key")
+	cmds, err := pipe.Exec(ctx)
+	assert.NoError(suite.T(), err)
+	assert.Equal(suite.T(), "value", cmds[1].(*redis.StringCmd).Val())
+}
+
+func (suite *PipelineTestSuite) TestTairPipelined() {
+	cmds, err := suite.tairClient.TairPipelined(ctx, func(p redis.Pipeliner) error {
+		p.Set(ctx, "key", "value", 0)
+		p.Get(ctx, "key")
+		return nil
+	})
+	assert.NoError(suite.T(), err)
+	assert.Equal(suite.T(), "value", cmds[1].(*redis.StringCmd).Val())
+}
+
+func (suite *PipelineTestSuite) TestTairClusterPipeline() {
+	pipe := suite.tairClusterClient.TairPipeline()
+	pipe.Set(ctx, "key", "value", 0)
+	pipe.Get(ctx, "key")
+	cmds, err := pipe.Exec(ctx)
+	assert.NoError(suite.T(), err)
+	assert.Equal(suite.T(), "value", cmds[1].(*redis.StringCmd).Val())
+}
+
+func (suite *PipelineTestSuite) TestTairClusterPipelined() {
+	cmds, err := suite.tairClusterClient.TairPipelined(ctx, func(p redis.Pipeliner) error {
+		p.Set(ctx, "key", "value", 0)
+		p.Get(ctx, "key")
+		return nil
+	})
+	assert.NoError(suite.T(), err)
+	assert.Equal(suite.T(), "value", cmds[1].(*redis.StringCmd).Val())
+}
+
+func TestTairPipelineTestSuite(t *testing.T) {
+	suite.Run(t, new(PipelineTestSuite))
+}