Support RequestPolicy

.github/workflows/modes/.env.enterprise

@@ -4,4 +4,4 @@ [email protected]
 RE_PASS=12345
 RE_CLUSTER_NAME=test
 RE_USE_OSS_CLUSTER=false
-RE_DB_PORT=6379
+RE_DB_PORT=6379

.gitignore

@@ -399,6 +399,7 @@ FodyWeavers.xsd
 .idea
 tests/NRedisStack.Tests/lcov.net7.0.info
 tests/NRedisStack.Tests/lcov.net6.0.info
+tests/NRedisStack.Tests/lcov.net8.0.info
 tests/NRedisStack.Tests/lcov.info
 tests/NRedisStack.Tests/.env
 tests/NRedisStack.Tests/redis_ca.pem

PackageVerification/PackageVerification.cs

@@ -24,7 +24,7 @@ public static void Main()
         ConnectionMultiplexer redis = ConnectionMultiplexer.Connect(configurationOptions);
 
         IDatabase db = redis.GetDatabase();
-        db.Execute("FLUSHALL");
+        db.FlushAll();
 
         IJsonCommands json = db.JSON();
 

src/NRedisStack/Auxiliary.cs

@@ -1,5 +1,8 @@
 using NRedisStack.Core;
+using NRedisStack.Core.Literals;
+using NRedisStack.Json.Literals;
 using NRedisStack.RedisStackCommands;
+using NRedisStack.Search.Literals;
 using StackExchange.Redis;
 
 namespace NRedisStack;
@@ -8,11 +11,27 @@ public static class Auxiliary
 {
     private static string? _libraryName = $"NRedisStack(.NET_v{Environment.Version})";
     private static bool _setInfo = true;
+    public static bool IsCluster(this IDatabase db) => db.Multiplexer.GetEndPoints().Length > 1;
+    public static bool IsEnterprise(this IDatabase db) // TODO: check if there is a better way to check if the server is Redis Enterprise without sending a command each time
+    {
+        // DPING command is available only in Redis Enterprise
+        try
+        {
+            db.Execute("DPING");
+            return true;
+        }
+        catch (RedisServerException)
+        {
+            return false;
+        }
+    }
+
     public static void ResetInfoDefaults()
     {
         _setInfo = true;
         _libraryName = $"NRedisStack(.NET_v{Environment.Version})";
     }
+
     public static List<object> MergeArgs(RedisKey key, params RedisValue[] items)
     {
         var args = new List<object>(items.Length + 1) { key };
@@ -36,7 +55,7 @@ public static object[] AssembleNonNullArguments(params object?[] arguments)
 
     // TODO: add all the signatures of GetDatabase
     public static IDatabase GetDatabase(this ConnectionMultiplexer redis,
-        string? LibraryName)
+                                        string? LibraryName)
     {
         var _db = redis.GetDatabase();
         if (LibraryName == null) // the user wants to disable the library name and version sending
@@ -61,66 +80,212 @@ internal static void SetInfoInPipeline(this IDatabase db)
         }
     }
 
+    // public static RedisResult Execute(this IDatabase db, SerializedCommand command)
+    // {
+    //     if (_setInfo)
+    //     {
+    //         _setInfo = false;
+    //         db.SetInfoInPipeline();
+    //     }
+    //     return (db.IsCluster()) ? db.ClusterExecute(command)
+    //                             : db.Execute(command.Command, command.Args);
+    // }
+
+    // public async static Task<RedisResult> ExecuteAsync(this IDatabaseAsync db, SerializedCommand command)
+    // {
+    //     if (_setInfo)
+    //     {
+    //         _setInfo = false;
+    //         ((IDatabase)db).SetInfoInPipeline();
+    //     }
+    //     return (db.IsCluster()) ? db.ClusterExecuteAsync(command)
+    //                             : db.ExecuteAsync(command.Command, command.Args);
+    // }
+
     public static RedisResult Execute(this IDatabase db, SerializedCommand command)
     {
         db.SetInfoInPipeline();
-        return db.Execute(command.Command, command.Args);
+
+        if (!db.IsEnterprise() || !db.IsCluster())
+            return db.Execute(command.Command, command.Args);
+
+        switch (command.Policy)
+        {
+            case RequestPolicy.Default: // add is cluster
+                return db.Execute(command.Command, command.Args);
+            case RequestPolicy.AllNodes:
+                return db.ExecuteAllNodes(command);
+            case RequestPolicy.AllShards:
+                return db.ExecuteAllShards(command);
+            case RequestPolicy.AnyShard:
+                return db.ExecuteAnyShard(command);
+            case RequestPolicy.MultiShard:
+                throw new NotImplementedException("MultiShard policy is not implemented yet");
+            case RequestPolicy.Special:
+                throw new NotImplementedException("Special policy is not implemented yet");
+            default:
+                throw new NotImplementedException("Unknown policy");
+        }
     }
 
     public static async Task<RedisResult> ExecuteAsync(this IDatabaseAsync db, SerializedCommand command)
     {
         ((IDatabase)db).SetInfoInPipeline();
-        return await db.ExecuteAsync(command.Command, command.Args);
+
+        if (!((IDatabase)db).IsCluster())
+            return await db.ExecuteAsync(command.Command, command.Args);
+
+        switch (command.Policy)
+        {
+            case RequestPolicy.Default:
+                return await db.ExecuteAsync(command.Command, command.Args);
+            case RequestPolicy.AllNodes:
+                return await db.ExecuteAllNodesAsync(command);
+            case RequestPolicy.AllShards:
+                return await db.ExecuteAllShardsAsync(command);
+            case RequestPolicy.AnyShard:
+                return await db.ExecuteAnyShardAsync(command);
+            case RequestPolicy.MultiShard:
+                // return db.ExecuteMultiShard(command);
+                throw new NotImplementedException("MultiShard policy is not implemented yet");
+            case RequestPolicy.Special:
+                throw new NotImplementedException("Special policy is not implemented yet");
+            default:
+                throw new NotImplementedException("Unknown policy");
+        }
     }
 
-    public static List<RedisResult> ExecuteBroadcast(this IDatabase db, string command)
-        => db.ExecuteBroadcast(new SerializedCommand(command));
+    public static RedisResult ExecuteAllNodes(this IDatabase db, SerializedCommand command)
+    {
+        var redis = db.Multiplexer;
+        var endpoints = redis.GetEndPoints();
+        var results = new RedisResult[endpoints.Length];
+
+        for (int i = 0; i < endpoints.Length; i++)
+        {
+            results[i] = redis.GetServer(endpoints[i]).Execute(command.Command, command.Args);
+        }
+
+        return results.ToRedisResult(command.Command);
+    }
 
-    public static List<RedisResult> ExecuteBroadcast(this IDatabase db, SerializedCommand command)
+    public static async Task<RedisResult> ExecuteAllNodesAsync(this IDatabaseAsync db, SerializedCommand command)
     {
         var redis = db.Multiplexer;
         var endpoints = redis.GetEndPoints();
-        var results = new List<RedisResult>(endpoints.Length);
+        var results = new RedisResult[endpoints.Length];
 
-        foreach (var endPoint in endpoints)
+        for (int i = 0; i < endpoints.Length; i++)
         {
-            var server = redis.GetServer(endPoint);
+            results[i] = await redis.GetServer(endpoints[i]).ExecuteAsync(command.Command, command.Args);
+        }
+
+        return results.ToRedisResult(command.Command);
+    }
+
+    public static RedisResult ExecuteAllShards(this IDatabase db, string command)
+            => db.ExecuteAllShards(new SerializedCommand(command));
+
+    public static RedisResult ExecuteAllShards(this IDatabase db, SerializedCommand command)
+    {
+        var redis = db.Multiplexer;
+        var endpoints = redis.GetEndPoints();
+        var results = new List<RedisResult>();
 
-            if (server.IsReplica)
+        foreach (var endpoint in endpoints)
+        {
+            var server = redis.GetServer(endpoint);
+            if (!server.IsReplica)
             {
-                continue; // Skip replica nodes
+                results.Add(server.Execute(command.Command, command.Args));
             }
-            // Send your command to the master node
-
-            results.Add(server.Multiplexer.GetDatabase().Execute(command));
         }
-        return results;
+
+        return results.ToArray().ToRedisResult(command.Command);
+
+
     }
 
-    public static async Task<List<RedisResult>> ExecuteBroadcastAsync(this IDatabaseAsync db, string command)
-        => await db.ExecuteBroadcastAsync(new SerializedCommand(command));
+    public async static Task<RedisResult> ExecuteAllShardsAsync(this IDatabaseAsync db, string command)
+            => await db.ExecuteAllShardsAsync(new SerializedCommand(command));
 
-    private static async Task<List<RedisResult>> ExecuteBroadcastAsync(this IDatabaseAsync db, SerializedCommand command)
+    public async static Task<RedisResult> ExecuteAllShardsAsync(this IDatabaseAsync db, SerializedCommand command)
     {
         var redis = db.Multiplexer;
         var endpoints = redis.GetEndPoints();
-        var results = new List<RedisResult>(endpoints.Length);
+        var results = new List<RedisResult>();
 
-        foreach (var endPoint in endpoints)
+        foreach (var endpoint in endpoints)
         {
-            var server = redis.GetServer(endPoint);
-
-            if (server.IsReplica)
+            var server = redis.GetServer(endpoint);
+            if (!server.IsReplica)
             {
-                continue; // Skip replica nodes
+                results.Add(await server.ExecuteAsync(command.Command, command.Args));
             }
-            // Send your command to the master node
+        }
+        var toRedisResult = results.ToArray().ToRedisResult(command.Command);
+
+        return toRedisResult;
+        return RedisResult.Create(results.ToArray());
+    }
+
+    public static RedisResult ExecuteAnyShard(this IDatabase db, SerializedCommand command)
+    {
+        var server = GetAnyPrimary(db.Multiplexer);
+        return server.Execute(command.Command, command.Args);
+    }
+
+    public static async Task<RedisResult> ExecuteAnyShardAsync(this IDatabaseAsync db, SerializedCommand command)
+    {
+        var server = GetAnyPrimary(db.Multiplexer);
+        return await server.ExecuteAsync(command.Command, command.Args);
+    }
+
+    public static IServer GetAnyPrimary(IConnectionMultiplexer muxer)
+    {
+        foreach (var endpoint in muxer.GetEndPoints())
+        {
+            var server = muxer.GetServer(endpoint);
+            if (!server.IsReplica) return server;
+        }
+        throw new InvalidOperationException("Requires a primary endpoint (found none)");
+    }
+
+    public static RedisResult ToRedisResult(this RedisResult[] results, string command)
+    {
+        switch (command)
+        {
+            case FT.ALIASADD:
+            case FT.ALIASDEL:
+            case FT.ALIASUPDATE:
+            case FT.ALTER:
+            case FT.CREATE:
+            case FT.DROPINDEX:
+            case FT.DICTADD:
+            case FT.DICTDEL:
+            case RedisCoreCommands.FLUSHALL:
+                return results.OKArraytoResult();
+            case JSON.MGET:
+                // TODO: implement
+                break;
+        }
+        return results[0]; // TODO: check if this is the correct behavior
+    }
 
-            results.Add(await server.Multiplexer.GetDatabase().ExecuteAsync(command));
+    public static RedisResult OKArraytoResult(this RedisResult[] results)
+    {
+        foreach (var result in results)
+        {
+            if (result.ToString() != "OK")
+            {
+                return result; // return the problematic result
+            }
         }
-        return results;
+        return results[0]; // return the first result (which is OK, like the others)
     }
 
+    // TODO: check if implementing MultiShard and Special policies is nessesary
+
     public static string GetNRedisStackVersion()
     {
         Version version = typeof(Auxiliary).Assembly.GetName().Version!;

src/NRedisStack/CoreCommands/CoreCommandBuilder.cs

@@ -199,5 +199,16 @@ private static SerializedCommand BlockingCommandWithKeysAndTimeout(String comman
 
             return new SerializedCommand(command, args);
         }
+
+        public static SerializedCommand FlushAll(bool? async)
+        {
+            List<object> args = new List<object>();
+            if (async != null)
+            {
+                args.Add(async.Value ? CoreArgs.ASYNC : CoreArgs.SYNC);
+            }
+
+            return new SerializedCommand(RedisCoreCommands.FLUSHALL, RequestPolicy.AllShards, args);
+        }
     }
 }

src/NRedisStack/CoreCommands/CoreCommands.cs

@@ -479,5 +479,18 @@ public static bool ClientSetInfo(this IDatabase db, SetInfoAttr attr, string val
         }
         return result[0].Entries;
     }
+
+    /// <summary>
+    /// Removes all keys from all databases.
+    /// </summary>
+    /// <param name="db">The <see cref="IDatabase"/> class where this extension method is applied.</param>
+    /// <param name="async">if set, flushes the databases asynchronously</param>
+    /// <returns><see langword="true"/> if everything was done correctly, Error otherwise.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/flushall"/></remarks>
+    public static bool FlushAll(this IDatabase db, bool? async = null)
+    {
+        var command = CoreCommandBuilder.FlushAll(async);
+        return db.Execute(command).OKtoBoolean();
+    }
 }
 

src/NRedisStack/CoreCommands/Literals/CommandArgs.cs

@@ -2,6 +2,7 @@ namespace NRedisStack.Core.Literals
 {
     internal static class CoreArgs
     {
+        public const string ASYNC = "ASYNC";
         public const string BLOCK = "BLOCK";
         public const string COUNT = "COUNT";
         public const string GROUP = "GROUP";
@@ -11,6 +12,7 @@ internal static class CoreArgs
         public const string NOACK = "NOACK";
         public const string RIGHT = "RIGHT";
         public const string STREAMS = "STREAMS";
+        public const string SYNC = "SYNC";
         public const string lib_name = "LIB-NAME";
         public const string lib_ver = "LIB-VER";
     }

src/NRedisStack/CoreCommands/Literals/Commands.cs

@@ -15,6 +15,7 @@ internal static class RedisCoreCommands
         public const string BZPOPMIN = "BZPOPMIN";
         public const string CLIENT = "CLIENT";
         public const string SETINFO = "SETINFO";
+        public const string FLUSHALL = "FLUSHALL";
         public const string XREAD = "XREAD";
         public const string XREADGROUP = "XREADGROUP";
     }

src/NRedisStack/Json/JsonCommandBuilder.cs

@@ -210,7 +210,7 @@ public static SerializedCommand MGet(RedisKey[] keys, string path)
         var args = keys.Cast<object>().ToList();
 
         args.Add(path);
-        return new SerializedCommand(JSON.MGET, args);
+        return new SerializedCommand(JSON.MGET, RequestPolicy.AllShards, args);
     }
 
     public static SerializedCommand NumIncrby(RedisKey key, string path, double value)