Merge pull request #85 from splitio/task/pipeline

Task/pipeline

CHANGES

@@ -1,3 +1,7 @@
+5.3.1 (Nov 29, 2023)
+ - Added Set, SAdd, SMembers, SRem, Incr, Decr, Del methods into Redis Pipeline.
+ - Updated mocks SAdd, SRem to accept members parameters.
+
 5.3.0 (March 8, 2023)
  - Upgraded redis/go-redis to v9.
  - Fixed golang.org/x/net vulnerability (CVE-2022-27664).

redis/mocks/mocks.go

@@ -72,12 +72,19 @@ func (m *MockResultOutput) MapStringString() (map[string]string, error) {
 
 // MpockPipeline  impl
 type MockPipeline struct {
-	LRangeCall  func(key string, start, stop int64)
-	LTrimCall   func(key string, start, stop int64)
-	LLenCall    func(key string)
-	HIncrByCall func(key string, field string, value int64)
-	HLenCall    func(key string)
-	ExecCall    func() ([]redis.Result, error)
+	LRangeCall   func(key string, start, stop int64)
+	LTrimCall    func(key string, start, stop int64)
+	LLenCall     func(key string)
+	HIncrByCall  func(key string, field string, value int64)
+	HLenCall     func(key string)
+	SetCall      func(key string, value interface{}, expiration time.Duration)
+	IncrCall     func(key string)
+	DecrCall     func(key string)
+	SAddCall     func(key string, members ...interface{})
+	SRemCall     func(key string, members ...interface{})
+	SMembersCall func(key string)
+	DelCall      func(keys ...string)
+	ExecCall     func() ([]redis.Result, error)
 }
 
 func (m *MockPipeline) LRange(key string, start, stop int64) {
@@ -100,6 +107,34 @@ func (m *MockPipeline) HLen(key string) {
 	m.HLenCall(key)
 }
 
+func (m *MockPipeline) Set(key string, value interface{}, expiration time.Duration) {
+	m.SetCall(key, value, expiration)
+}
+
+func (m *MockPipeline) Incr(key string) {
+	m.IncrCall(key)
+}
+
+func (m *MockPipeline) Decr(key string) {
+	m.DecrCall(key)
+}
+
+func (m *MockPipeline) SAdd(key string, members ...interface{}) {
+	m.SAddCall(key, members...)
+}
+
+func (m *MockPipeline) SRem(key string, members ...interface{}) {
+	m.SRemCall(key, members...)
+}
+
+func (m *MockPipeline) SMembers(key string) {
+	m.SMembersCall(key)
+}
+
+func (m *MockPipeline) Del(keys ...string) {
+	m.DelCall(keys...)
+}
+
 func (m *MockPipeline) Exec() ([]redis.Result, error) {
 	return m.ExecCall()
 }
@@ -196,12 +231,12 @@ func (m *MockClient) SIsMember(key string, member interface{}) redis.Result {
 
 // SAdd mocks SAdd
 func (m *MockClient) SAdd(key string, members ...interface{}) redis.Result {
-	return m.SAddCall(key)
+	return m.SAddCall(key, members...)
 }
 
 // SRem mocks SRem
 func (m *MockClient) SRem(key string, members ...interface{}) redis.Result {
-	return m.SRemCall(key)
+	return m.SRemCall(key, members...)
 }
 
 // Incr mocks Incr

redis/prefixedclient.go

@@ -233,6 +233,45 @@ func (p *PrefixedPipeline) HLen(key string) {
 	p.wrapped.HLen(withPrefix(p.prefix, key))
 }
 
+// Set schedules a Set operation on this pipeline
+func (p *PrefixedPipeline) Set(key string, value interface{}, expiration time.Duration) {
+	p.wrapped.Set(withPrefix(p.prefix, key), value, expiration)
+}
+
+// Incr schedules an Incr operation on this pipeline
+func (p *PrefixedPipeline) Incr(key string) {
+	p.wrapped.Incr(withPrefix(p.prefix, key))
+}
+
+// Decr schedules a Decr operation on this pipeline
+func (p *PrefixedPipeline) Decr(key string) {
+	p.wrapped.Decr(withPrefix(p.prefix, key))
+}
+
+// SAdd schedules a SAdd operation on this pipeline
+func (p *PrefixedPipeline) SAdd(key string, members ...interface{}) {
+	p.wrapped.SAdd(withPrefix(p.prefix, key), members...)
+}
+
+// SRem schedules a SRem operation on this pipeline
+func (p *PrefixedPipeline) SRem(key string, members ...interface{}) {
+	p.wrapped.SRem(withPrefix(p.prefix, key), members...)
+}
+
+// SMembers schedules a SMembers operation on this pipeline
+func (p *PrefixedPipeline) SMembers(key string) {
+	p.wrapped.SMembers(withPrefix(p.prefix, key))
+}
+
+// Del schedules a Del operation on this pipeline
+func (p *PrefixedPipeline) Del(keys ...string) {
+	prefixedKeys := make([]string, len(keys))
+	for i, k := range keys {
+		prefixedKeys[i] = withPrefix(p.prefix, k)
+	}
+	p.wrapped.Del(prefixedKeys...)
+}
+
 // Exec executes the pipeline
 func (p *PrefixedPipeline) Exec() ([]Result, error) {
 	return p.wrapped.Exec()

redis/wrapper.go

@@ -95,6 +95,13 @@ type Pipeline interface {
 	LLen(key string)
 	HIncrBy(key string, field string, value int64)
 	HLen(key string)
+	Set(key string, value interface{}, expiration time.Duration)
+	Incr(key string)
+	Decr(key string)
+	SAdd(key string, members ...interface{})
+	SRem(key string, members ...interface{})
+	SMembers(key string)
+	Del(keys ...string)
 	Exec() ([]Result, error)
 }
 
@@ -128,6 +135,41 @@ func (p *PipelineImpl) HLen(key string) {
 	p.wrapped.HLen(context.TODO(), key)
 }
 
+// Set schedules a Set operation on this pipeline
+func (p *PipelineImpl) Set(key string, value interface{}, expiration time.Duration) {
+	p.wrapped.Set(context.TODO(), key, value, expiration)
+}
+
+// Incr schedules an Incr operation on this pipeline
+func (p *PipelineImpl) Incr(key string) {
+	p.wrapped.Incr(context.TODO(), key)
+}
+
+// Decr schedules a Decr operation on this pipeline
+func (p *PipelineImpl) Decr(key string) {
+	p.wrapped.Decr(context.TODO(), key)
+}
+
+// SAdd schedules a SAdd operation on this pipeline
+func (p *PipelineImpl) SAdd(key string, members ...interface{}) {
+	p.wrapped.SAdd(context.TODO(), key, members...)
+}
+
+// SRem schedules a SRem operation on this pipeline
+func (p *PipelineImpl) SRem(key string, members ...interface{}) {
+	p.wrapped.SRem(context.TODO(), key, members...)
+}
+
+// SMembers schedules a SMembers operation on this pipeline
+func (p *PipelineImpl) SMembers(key string) {
+	p.wrapped.SMembers(context.TODO(), key)
+}
+
+// Del schedules a Del operation on this pipeline
+func (p *PipelineImpl) Del(keys ...string) {
+	p.wrapped.Del(context.TODO(), keys...)
+}
+
 // Exec executes the pipeline
 func (p *PipelineImpl) Exec() ([]Result, error) {
 	res, err := p.wrapped.Exec(context.TODO())

redis/wrapper_test.go

@@ -2,6 +2,7 @@ package redis
 
 import (
 	"testing"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 	"github.com/splitio/go-toolkit/v5/testhelpers"
@@ -13,7 +14,10 @@ func TestRedisWrapperPipeline(t *testing.T) {
 
 	client.Del("key1")
 	client.Del("key-test")
+	client.Del("key-set")
 	client.RPush("key1", "e1", "e2", "e3")
+	client.Set("key-del1", 0, 1*time.Hour)
+	client.Set("key-del2", 0, 1*time.Hour)
 
 	pipe := client.Pipeline()
 	pipe.LRange("key1", 0, 5)
@@ -23,13 +27,20 @@ func TestRedisWrapperPipeline(t *testing.T) {
 	pipe.HIncrBy("key-test", "field-test-2", 4)
 	pipe.HIncrBy("key-test", "field-test-2", 3)
 	pipe.HLen("key-test")
+	pipe.Set("key-set", "field-test-1", 1*time.Hour)
+	pipe.SAdd("key-sadd", []interface{}{"field-test-1", "field-test-2"})
+	pipe.SMembers("key-sadd")
+	pipe.SRem("key-sadd", []interface{}{"field-test-1", "field-test-2"})
+	pipe.Incr("key-incr")
+	pipe.Decr("key-incr")
+	pipe.Del([]string{"key-del1", "key-del2"}...)
 	result, err := pipe.Exec()
 	if err != nil {
 		t.Error("there should not be any error. Got: ", err)
 	}
 
-	if len(result) != 7 {
-		t.Error("there should be 4 elements")
+	if len(result) != 14 {
+		t.Error("there should be 13 elements")
 	}
 
 	items, _ := result[0].Multi()
@@ -61,4 +72,27 @@ func TestRedisWrapperPipeline(t *testing.T) {
 	if ib := client.HIncrBy("key-test", "field-test", 1); ib.Int() != 6 {
 		t.Error("new count should be 6")
 	}
+
+	if ib := client.Get("key-set"); ib.String() != "field-test-1" {
+		t.Error("it should be field-test-1")
+	}
+
+	if c := result[8].Int(); c != 2 {
+		t.Error("count should be 2. Is: ", c)
+	}
+	if d, _ := result[9].Multi(); len(d) != 2 {
+		t.Error("count should be 2. Is: ", len(d))
+	}
+	if c := result[10].Int(); c != 2 {
+		t.Error("count should be 2. Is: ", c)
+	}
+	if c := result[11].Int(); c != 1 {
+		t.Error("count should be 1. Is: ", c)
+	}
+	if c := result[12].Int(); c != 0 {
+		t.Error("count should be zero. Is: ", c)
+	}
+	if c := result[13].Int(); c != 2 {
+		t.Error("count should be 2. Is: ", c)
+	}
 }