fix(redis): Implement redisNode interface for redis.Redis

core/stores/redis/redis.go

@@ -90,6 +90,8 @@ type (
 	StringCmd = red.StringCmd
 	// Script is an alias of redis.Script.
 	Script = red.Script
+	// CommandsInfoCmd is an alias of redis.CommandsInfoCmd.
+	CommandsInfoCmd = red.CommandsInfoCmd
 
 	// Hook is an alias of redis.Hook.
 	Hook = red.Hook
@@ -243,6 +245,16 @@ func (s *Redis) BitOpXorCtx(ctx context.Context, destKey string, keys ...string)
 	return conn.BitOpXor(ctx, destKey, keys...).Result()
 }
 
+// Command is the implementation of redis command command.
+func (s *Redis) Command(ctx context.Context) *CommandsInfoCmd {
+	conn, err := getRedis(s)
+	if err != nil {
+		return nil
+	}
+
+	return conn.Command(ctx)
+}
+
 // BitPos is redis bitpos command implementation.
 func (s *Redis) BitPos(key string, bit, start, end int64) (int64, error) {
 	return s.BitPosCtx(context.Background(), key, bit, start, end)
@@ -1401,6 +1413,26 @@ func (s *Redis) ScriptLoadCtx(ctx context.Context, script string) (string, error
 	return conn.ScriptLoad(ctx, script).Result()
 }
 
+// TxPipelined is the implementation of redis txpipelined command.
+func (s *Redis) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
+	conn, err := getRedis(s)
+	if err != nil {
+		return nil, err
+	}
+
+	return conn.TxPipelined(ctx, fn)
+}
+
+// Pipeline is the implementation of redis pipeline command.
+func (s *Redis) Pipeline() Pipeliner {
+	conn, err := getRedis(s)
+	if err != nil {
+		return nil
+	}
+
+	return conn.Pipeline()
+}
+
 // ScriptRun is the implementation of *redis.Script run command.
 func (s *Redis) ScriptRun(script *Script, keys []string, args ...any) (any, error) {
 	return s.ScriptRunCtx(context.Background(), script, keys, args...)

core/stores/redis/redis_test.go

@@ -2172,38 +2172,73 @@ func TestRedisUnlink(t *testing.T) {
 	})
 }
 
-func TestRedisTxPipeline(t *testing.T) {
+func TestRedis_Pipeline(t *testing.T) {
 	runOnRedis(t, func(client *Redis) {
 		ctx := context.Background()
-		_, err := newRedis(client.Addr, badType()).TxPipeline()
-		assert.NotNil(t, err)
-		pipe, err := client.TxPipeline()
-		assert.Nil(t, err)
+		pipe := client.Pipeline()
+		assert.NotNil(t, pipe)
+
 		key := "key"
-		hashKey := "field"
-		hashValue := "value"
+		value := "value"
 
-		// setting value
-		pipe.HSet(ctx, key, hashKey, hashValue)
+		pipe.Set(ctx, key, value, 0)
+		getCmd := pipe.Get(ctx, key)
 
-		existsCmd := pipe.Exists(ctx, key)
-		getCmd := pipe.HGet(ctx, key, hashKey)
+		_, err := pipe.Exec(ctx)
+		assert.NoError(t, err)
 
-		// execution
-		_, err = pipe.Exec(ctx)
-		assert.Nil(t, err)
+		val, err := getCmd.Result()
+		assert.NoError(t, err)
+		assert.Equal(t, value, val)
+	})
+}
 
-		// verification results
-		exists, err := existsCmd.Result()
-		assert.Nil(t, err)
-		assert.Equal(t, int64(1), exists)
+func TestRedis_Pipeline_Error(t *testing.T) {
+	r := newRedis("localhost:6379", badType())
+	pipe := r.Pipeline()
+	assert.Nil(t, pipe)
+}
 
-		value, err := getCmd.Result()
-		assert.Nil(t, err)
-		assert.Equal(t, hashValue, value)
+func TestRedis_TxPipelined(t *testing.T) {
+	runOnRedis(t, func(client *Redis) {
+		ctx := context.Background()
+		var incr *red.IntCmd
+		fn := func(p red.Pipeliner) error {
+			incr = p.Incr(ctx, "key")
+			return nil
+		}
+
+		_, err := client.TxPipelined(ctx, fn)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(1), incr.Val())
+	})
+}
+
+func TestRedis_TxPipelined_Error(t *testing.T) {
+	r := newRedis("localhost:6379", badType())
+	_, err := r.TxPipelined(context.Background(), func(p red.Pipeliner) error {
+		return nil
+	})
+	assert.Error(t, err)
+}
+
+func TestRedis_Command(t *testing.T) {
+	runOnRedis(t, func(client *Redis) {
+		ctx := context.Background()
+		cmd := client.Command(ctx)
+		assert.NotNil(t, cmd)
+		val, err := cmd.Result()
+		assert.NoError(t, err)
+		assert.NotEmpty(t, val)
 	})
 }
 
+func TestRedis_Command_Error(t *testing.T) {
+	r := newRedis("localhost:6379", badType())
+	cmd := r.Command(context.Background())
+	assert.Nil(t, cmd)
+}
+
 func TestRedisXGroupCreate(t *testing.T) {
 	runOnRedis(t, func(client *Redis) {
 		_, err := newRedis(client.Addr, badType()).XGroupCreate("Source", "Destination", "0")