X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=internal%2Fsdlgoredis%2Fsdlgoredis.go;h=8e2a2ef2e12e265dfce2b6f17a3adab5cfacc9b1;hb=refs%2Ftags%2Fv0.5.0;hp=b032d8f566c70e9699355c7f872175731578386b;hpb=135580f96b31e8dd0ccd5695df877a7942b912ae;p=ric-plt%2Fsdlgo.git

diff --git a/internal/sdlgoredis/sdlgoredis.go b/internal/sdlgoredis/sdlgoredis.go
index b032d8f..8e2a2ef 100644
--- a/internal/sdlgoredis/sdlgoredis.go
+++ b/internal/sdlgoredis/sdlgoredis.go
@@ -21,7 +21,9 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/go-redis/redis"
 )
@@ -35,12 +37,44 @@ type intChannels struct {
 }
 
 type DB struct {
-	client       *redis.Client
+	client       RedisClient
+	subscribe    SubscribeFn
 	redisModules bool
 	cbMap        map[string]ChannelNotificationCb
 	ch           intChannels
 }
 
+type Subscriber interface {
+	Channel() <-chan *redis.Message
+	Subscribe(channels ...string) error
+	Unsubscribe(channels ...string) error
+	Close() error
+}
+
+type SubscribeFn func(client RedisClient, channels ...string) Subscriber
+
+type RedisClient interface {
+	Command() *redis.CommandsInfoCmd
+	Close() error
+	Subscribe(channels ...string) *redis.PubSub
+	MSet(pairs ...interface{}) *redis.StatusCmd
+	Do(args ...interface{}) *redis.Cmd
+	MGet(keys ...string) *redis.SliceCmd
+	Del(keys ...string) *redis.IntCmd
+	Keys(pattern string) *redis.StringSliceCmd
+	SetNX(key string, value interface{}, expiration time.Duration) *redis.BoolCmd
+	SAdd(key string, members ...interface{}) *redis.IntCmd
+	SRem(key string, members ...interface{}) *redis.IntCmd
+	SMembers(key string) *redis.StringSliceCmd
+	SIsMember(key string, member interface{}) *redis.BoolCmd
+	SCard(key string) *redis.IntCmd
+	PTTL(key string) *redis.DurationCmd
+	Eval(script string, keys []string, args ...interface{}) *redis.Cmd
+	EvalSha(sha1 string, keys []string, args ...interface{}) *redis.Cmd
+	ScriptExists(scripts ...string) *redis.BoolSliceCmd
+	ScriptLoad(script string) *redis.StringCmd
+}
+
 func checkResultAndError(result interface{}, err error) (bool, error) {
 	if err != nil {
 		if err == redis.Nil {
@@ -50,42 +84,28 @@ func checkResultAndError(result interface{}, err error) (bool, error) {
 	}
 	if result == "OK" {
 		return true, nil
-	} else {
-		return false, nil
 	}
+	return false, nil
 }
 
 func checkIntResultAndError(result interface{}, err error) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	if result.(int64) == 1 {
+	if result.(int64) == int64(1) {
 		return true, nil
-	} else {
-		return false, nil
 	}
-
+	return false, nil
 }
 
-func Create() *DB {
-	hostname := os.Getenv("DBAAS_SERVICE_HOST")
-	if hostname == "" {
-		hostname = "localhost"
-	}
-	port := os.Getenv("DBAAS_SERVICE_PORT")
-	if port == "" {
-		port = "6379"
-	}
-	redisAddress := hostname + ":" + port
-	client := redis.NewClient(&redis.Options{
-		Addr:     redisAddress,
-		Password: "", // no password set
-		DB:       0,  // use default DB
-		PoolSize: 20,
-	})
+func subscribeNotifications(client RedisClient, channels ...string) Subscriber {
+	return client.Subscribe(channels...)
+}
 
+func CreateDB(client RedisClient, subscribe SubscribeFn) *DB {
 	db := DB{
 		client:       client,
+		subscribe:    subscribe,
 		redisModules: true,
 		cbMap:        make(map[string]ChannelNotificationCb, 0),
 		ch: intChannels{
@@ -95,9 +115,49 @@ func Create() *DB {
 		},
 	}
 
+	return &db
+}
+
+func Create() *DB {
+	var client *redis.Client
+	hostname := os.Getenv("DBAAS_SERVICE_HOST")
+	if hostname == "" {
+		hostname = "localhost"
+	}
+	port := os.Getenv("DBAAS_SERVICE_PORT")
+	if port == "" {
+		port = "6379"
+	}
+	sentinelPort := os.Getenv("DBAAS_SERVICE_SENTINEL_PORT")
+	masterName := os.Getenv("DBAAS_MASTER_NAME")
+	if sentinelPort == "" {
+		redisAddress := hostname + ":" + port
+		client = redis.NewClient(&redis.Options{
+			Addr:       redisAddress,
+			Password:   "", // no password set
+			DB:         0,  // use default DB
+			PoolSize:   20,
+			MaxRetries: 2,
+		})
+	} else {
+		sentinelAddress := hostname + ":" + sentinelPort
+		client = redis.NewFailoverClient(&redis.FailoverOptions{
+			MasterName:    masterName,
+			SentinelAddrs: []string{sentinelAddress},
+			PoolSize:      20,
+			MaxRetries:    2,
+		})
+	}
+	db := CreateDB(client, subscribeNotifications)
+	db.CheckCommands()
+	return db
+}
+
+func (db *DB) CheckCommands() {
 	commands, err := db.client.Command().Result()
 	if err == nil {
-		redisModuleCommands := []string{"setie", "delie", "msetpub", "setiepub", "setnxpub", "delpub"}
+		redisModuleCommands := []string{"setie", "delie", "setiepub", "setnxpub",
+			"msetmpub", "delmpub"}
 		for _, v := range redisModuleCommands {
 			_, ok := commands[v]
 			if !ok {
@@ -107,7 +167,6 @@ func Create() *DB {
 	} else {
 		fmt.Println(err)
 	}
-	return &db
 }
 
 func (db *DB) CloseDB() error {
@@ -124,7 +183,7 @@ func (db *DB) UnsubscribeChannelDB(channels ...string) {
 	}
 }
 
-func (db *DB) SubscribeChannelDB(cb ChannelNotificationCb, channelPrefix, eventSeparator string, channels ...string) {
+func (db *DB) SubscribeChannelDB(cb func(string, ...string), channelPrefix, eventSeparator string, channels ...string) {
 	if len(db.cbMap) == 0 {
 		for _, v := range channels {
 			db.cbMap[v] = cb
@@ -135,7 +194,7 @@ func (db *DB) SubscribeChannelDB(cb ChannelNotificationCb, channelPrefix, eventS
 			eventSeparator string,
 			ch intChannels,
 			channels ...string) {
-			sub := db.client.Subscribe(channels...)
+			sub := db.subscribe(db.client, channels...)
 			rxChannel := sub.Channel()
 			for {
 				select {
@@ -171,16 +230,20 @@ func (db *DB) MSet(pairs ...interface{}) error {
 	return db.client.MSet(pairs...).Err()
 }
 
-func (db *DB) MSetPub(channel, message string, pairs ...interface{}) error {
+func (db *DB) MSetMPub(channelsAndEvents []string, pairs ...interface{}) error {
 	if !db.redisModules {
-		return errors.New("Redis deployment doesn't support MSETPUB command")
+		return errors.New("Redis deployment doesn't support MSETMPUB command")
 	}
 	command := make([]interface{}, 0)
-	command = append(command, "MSETPUB")
+	command = append(command, "MSETMPUB")
+	command = append(command, len(pairs)/2)
+	command = append(command, len(channelsAndEvents)/2)
 	for _, d := range pairs {
 		command = append(command, d)
 	}
-	command = append(command, channel, message)
+	for _, d := range channelsAndEvents {
+		command = append(command, d)
+	}
 	_, err := db.client.Do(command...).Result()
 	return err
 }
@@ -189,18 +252,23 @@ func (db *DB) MGet(keys []string) ([]interface{}, error) {
 	return db.client.MGet(keys...).Result()
 }
 
-func (db *DB) DelPub(channel, message string, keys []string) error {
+func (db *DB) DelMPub(channelsAndEvents []string, keys []string) error {
 	if !db.redisModules {
-		return errors.New("Redis deployment not supporting command DELPUB")
+		return errors.New("Redis deployment not supporting command DELMPUB")
 	}
 	command := make([]interface{}, 0)
-	command = append(command, "DELPUB")
+	command = append(command, "DELMPUB")
+	command = append(command, len(keys))
+	command = append(command, len(channelsAndEvents)/2)
 	for _, d := range keys {
 		command = append(command, d)
 	}
-	command = append(command, channel, message)
+	for _, d := range channelsAndEvents {
+		command = append(command, d)
+	}
 	_, err := db.client.Do(command...).Result()
 	return err
+
 }
 
 func (db *DB) Del(keys []string) error {
@@ -233,8 +301,8 @@ func (db *DB) SetNXPub(channel, message, key string, data interface{}) (bool, er
 	}
 	return checkResultAndError(db.client.Do("SETNXPUB", key, data, channel, message).Result())
 }
-func (db *DB) SetNX(key string, data interface{}) (bool, error) {
-	return db.client.SetNX(key, data, 0).Result()
+func (db *DB) SetNX(key string, data interface{}, expiration time.Duration) (bool, error) {
+	return db.client.SetNX(key, data, expiration).Result()
 }
 
 func (db *DB) DelIEPub(channel, message, key string, data interface{}) (bool, error) {
@@ -275,3 +343,22 @@ func (db *DB) SCard(key string) (int64, error) {
 	result, err := db.client.SCard(key).Result()
 	return result, err
 }
+
+func (db *DB) PTTL(key string) (time.Duration, error) {
+	result, err := db.client.PTTL(key).Result()
+	return result, err
+}
+
+var luaRefresh = redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("pexpire", KEYS[1], ARGV[2]) else return 0 end`)
+
+func (db *DB) PExpireIE(key string, data interface{}, expiration time.Duration) error {
+	expirationStr := strconv.FormatInt(int64(expiration/time.Millisecond), 10)
+	result, err := luaRefresh.Run(db.client, []string{key}, data, expirationStr).Result()
+	if err != nil {
+		return err
+	}
+	if result == int64(1) {
+		return nil
+	}
+	return errors.New("Lock not held")
+}