Merge branch 'master' into marc/editorconfig

Author: mgravell

.github/workflows/integration.yml

@@ -37,7 +37,7 @@ jobs:
         max-parallel: 15
         fail-fast: false
         matrix:
-          redis-version: [ '8.2-rc2-pre', '${{ needs.redis_version.outputs.CURRENT }}', '7.4.1', '7.2.6', '6.2.16']
+          redis-version: [ '8.2.1-pre', '${{ needs.redis_version.outputs.CURRENT }}', '7.4.1', '7.2.6', '6.2.16']
           dotnet-version: ['8.0', '9.0']
       env:
         ACTIONS_ALLOW_UNSECURE_COMMANDS: true

.gitignore

@@ -410,3 +410,8 @@ tests/NRedisStack.Tests/redis_credentials/redis_user.crt
 # global.json
 global.json
 tests/NRedisStack.Tests/lcov.net8.0.info
+
+# docker containers
+tests/dockers/cluster/
+tests/dockers/standalone/
+tests/dockers/all/

Directory.Packages.props

@@ -12,7 +12,7 @@
     <!-- primary library -->
     <PackageVersion Include="NetTopologySuite" Version="2.6.0" />
     <PackageVersion Include="System.Text.Json" Version="9.0.7" />
-    <PackageVersion Include="StackExchange.Redis" Version="2.8.58" />
+    <PackageVersion Include="StackExchange.Redis" Version="2.9.17" />
     <!-- tests, etc -->
     <PackageVersion Include="BouncyCastle.Cryptography" Version="2.6.1" />
     <PackageVersion Include="coverlet.collector" Version="6.0.4" />

src/NRedisStack/Auxiliary.cs

@@ -67,12 +67,22 @@ public static RedisResult Execute(this IDatabase db, SerializedCommand command)
         return db.Execute(command.Command, command.Args);
     }
 
+    internal static RedisResult Execute(this IServer server, int? db, SerializedCommand command)
+    {
+        return server.Execute(db, command.Command, command.Args);
+    }
+
     public static async Task<RedisResult> ExecuteAsync(this IDatabaseAsync db, SerializedCommand command)
     {
         ((IDatabase)db).SetInfoInPipeline();
         return await db.ExecuteAsync(command.Command, command.Args);
     }
 
+    internal static async Task<RedisResult> ExecuteAsync(this IServer server, int? db, SerializedCommand command)
+    {
+        return await server.ExecuteAsync(db, command.Command, command.Args);
+    }
+
     public static List<RedisResult> ExecuteBroadcast(this IDatabase db, string command)
         => db.ExecuteBroadcast(new SerializedCommand(command));
 

src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt

@@ -1 +1,13 @@
 #nullable enable
+NRedisStack.ISearchCommands.AggregateEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.ISearchCommands.CursorDel(NRedisStack.Search.AggregationResult! result) -> bool
+NRedisStack.ISearchCommands.CursorRead(NRedisStack.Search.AggregationResult! result, int? count = null) -> NRedisStack.Search.AggregationResult!
+NRedisStack.ISearchCommandsAsync.AggregateAsyncEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IAsyncEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.ISearchCommandsAsync.CursorDelAsync(NRedisStack.Search.AggregationResult! result) -> System.Threading.Tasks.Task<bool>!
+NRedisStack.ISearchCommandsAsync.CursorReadAsync(NRedisStack.Search.AggregationResult! result, int? count = null) -> System.Threading.Tasks.Task<NRedisStack.Search.AggregationResult!>!
+NRedisStack.SearchCommands.AggregateEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.SearchCommands.CursorDel(NRedisStack.Search.AggregationResult! result) -> bool
+NRedisStack.SearchCommands.CursorRead(NRedisStack.Search.AggregationResult! result, int? count = null) -> NRedisStack.Search.AggregationResult!
+NRedisStack.SearchCommandsAsync.AggregateAsyncEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IAsyncEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.SearchCommandsAsync.CursorDelAsync(NRedisStack.Search.AggregationResult! result) -> System.Threading.Tasks.Task<bool>!
+NRedisStack.SearchCommandsAsync.CursorReadAsync(NRedisStack.Search.AggregationResult! result, int? count = null) -> System.Threading.Tasks.Task<NRedisStack.Search.AggregationResult!>!

src/NRedisStack/ResponseParser.cs

@@ -737,6 +737,20 @@ public static AggregationResult ToAggregationResult(this RedisResult result, Agg
         }
     }
 
+    internal static AggregationResult ToAggregationResult(this RedisResult result, string indexName, AggregationRequest query, IServer? server, int? database)
+    {
+        if (query.IsWithCursor())
+        {
+            var results = (RedisResult[])result!;
+
+            return new AggregationResult.WithCursorAggregationResult(indexName, results[0], (long)results[1], server, database);
+        }
+        else
+        {
+            return new(result);
+        }
+    }
+
     public static Dictionary<string, RedisResult>[] ToDictionarys(this RedisResult result)
     {
         var resArr = (RedisResult[])result!;

src/NRedisStack/Search/AggregationRequest.cs

@@ -128,6 +128,7 @@ public AggregationRequest Cursor(int? count = null, long? maxIdle = null)
 
         if (count != null)
         {
+            Count = count;
             args.Add(SearchArgs.COUNT);
             args.Add(count);
         }
@@ -139,6 +140,7 @@ public AggregationRequest Cursor(int? count = null, long? maxIdle = null)
         }
         return this;
     }
+    internal int? Count { get; set; }
 
     public AggregationRequest Params(Dictionary<string, object> nameValue)
     {

src/NRedisStack/Search/AggregationResult.cs

@@ -3,15 +3,29 @@
 
 namespace NRedisStack.Search;
 
-public sealed class AggregationResult
+public class AggregationResult
 {
+    // internal subclass for WITHCURSOR calls, which need to be issued to the same connection
+    internal sealed class WithCursorAggregationResult : AggregationResult
+    {
+        internal WithCursorAggregationResult(string indexName, RedisResult result, long cursorId, IServer? server,
+            int? database) : base(result, cursorId)
+        {
+            IndexName = indexName;
+            Server = server;
+            Database = database;
+        }
+        public string IndexName { get; }
+        public IServer? Server { get; }
+        public int? Database { get; }
+    }
+
     public long TotalResults { get; }
     private readonly Dictionary<string, object>[] _results;
     private Dictionary<string, RedisValue>[]? _resultsAsRedisValues;
 
     public long CursorId { get; }
 
-
     internal AggregationResult(RedisResult result, long cursorId = -1)
     {
         var arr = (RedisResult[])result!;
@@ -45,7 +59,6 @@ internal AggregationResult(RedisResult result, long cursorId = -1)
         CursorId = cursorId;
     }
 
-
     /// <summary>
     /// takes a Redis multi-bulk array represented by a RedisResult[] and recursively processes its elements.
     /// For each element in the array, it checks if it's another multi-bulk array, and if so, it recursively calls itself.

src/NRedisStack/Search/ISearchCommands.cs

@@ -1,4 +1,6 @@
+using System.ComponentModel;
 using NRedisStack.Search;
+using NRedisStack.Search.Aggregation;
 using NRedisStack.Search.DataTypes;
 using StackExchange.Redis;
 
@@ -18,11 +20,20 @@ public interface ISearchCommands
     /// Run a search query on an index, and perform aggregate transformations on the results.
     /// </summary>
     /// <param name="index">The index name.</param>
-    /// <param name="query">The query</param>
+    /// <param name="query">The query.</param>
     /// <returns>An <see langword="AggregationResult"/> object</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
     AggregationResult Aggregate(string index, AggregationRequest query);
 
+    /// <summary>
+    /// Run a search query on an index, and perform aggregate transformations on the results.
+    /// </summary>
+    /// <param name="index">The index name.</param>
+    /// <param name="query">The query.</param>
+    /// <returns>A sequence of <see langword="Row"/> values.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
+    IEnumerable<Row> AggregateEnumerable(string index, AggregationRequest query);
+
     /// <summary>
     /// Add an alias to an index.
     /// </summary>
@@ -92,22 +103,43 @@ public interface ISearchCommands
     /// <summary>
     /// Delete a cursor from the index.
     /// </summary>
-    /// <param name="indexName">The index name</param>
+    /// <param name="indexName">The index name.</param>
     /// <param name="cursorId">The cursor's ID.</param>
     /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    [Obsolete("When possible, use CursorDel(AggregationResult) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     bool CursorDel(string indexName, long cursorId);
 
+    /// <summary>
+    /// Delete a cursor from the index.
+    /// </summary>
+    /// <param name="result">The result of a previous call to Aggregate or CursorRead.</param>
+    /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    bool CursorDel(AggregationResult result);
+
     /// <summary>
     /// Read next results from an existing cursor.
     /// </summary>
-    /// <param name="indexName">The index name</param>
+    /// <param name="indexName">The index name.</param>
     /// <param name="cursorId">The cursor's ID.</param>
     /// <param name="count">Limit the amount of returned results.</param>
     /// <returns>A AggregationResult object with the results</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    [Obsolete("When possible, use AggregateEnumerable or CursorRead(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     AggregationResult CursorRead(string indexName, long cursorId, int? count = null);
 
+    /// <summary>
+    /// Read next results from an existing cursor.
+    /// </summary>
+    /// <param name="result">The result of a previous call to Aggregate or CursorRead.</param>
+    /// <param name="count">Limit the amount of returned results.</param>
+    /// <returns>A AggregationResult object with the results</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    public AggregationResult CursorRead(AggregationResult result, int? count = null);
+
     /// <summary>
     /// Add terms to a dictionary.
     /// </summary>

src/NRedisStack/Search/ISearchCommandsAsync.cs

@@ -1,4 +1,6 @@
+using System.ComponentModel;
 using NRedisStack.Search;
+using NRedisStack.Search.Aggregation;
 using NRedisStack.Search.DataTypes;
 using StackExchange.Redis;
 
@@ -14,14 +16,25 @@ public interface ISearchCommandsAsync
     Task<RedisResult[]> _ListAsync();
 
     /// <summary>
-    /// Run a search query on an index, and perform aggregate transformations on the results.
+    /// Run a search query on an index, and perform aggregate transformations on the results. This operates
+    /// as a cursor and may involve multiple commands to the server.
     /// </summary>
     /// <param name="index">The index name.</param>
     /// <param name="query">The query</param>
     /// <returns>An <see langword="AggregationResult"/> object</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
     Task<AggregationResult> AggregateAsync(string index, AggregationRequest query);
 
+    /// <summary>
+    /// Run a search query on an index, and perform aggregate transformations on the results. This operates
+    /// as a cursor and may involve multiple commands to the server.
+    /// </summary>
+    /// <param name="index">The index name.</param>
+    /// <param name="query">The query.</param>
+    /// <returns>A sequence of <see langword="Row"/> values.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
+    IAsyncEnumerable<Row> AggregateAsyncEnumerable(string index, AggregationRequest query);
+
     /// <summary>
     /// Add an alias to an index.
     /// </summary>
@@ -95,8 +108,18 @@ public interface ISearchCommandsAsync
     /// <param name="cursorId">The cursor's ID.</param>
     /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    [Obsolete("When possible, use CursorDelAsync(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     Task<bool> CursorDelAsync(string indexName, long cursorId);
 
+    /// <summary>
+    /// Delete a cursor from the index.
+    /// </summary>
+    /// <param name="result">The result of a previous call to AggregateAsync or CursorReadAsync.</param>
+    /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    Task<bool> CursorDelAsync(AggregationResult result);
+
     /// <summary>
     /// Read next results from an existing cursor.
     /// </summary>
@@ -105,8 +128,19 @@ public interface ISearchCommandsAsync
     /// <param name="count">Limit the amount of returned results.</param>
     /// <returns>A AggregationResult object with the results</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    [Obsolete("When possible, use AggregateAsyncEnumerable or CursorReadAsync(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     Task<AggregationResult> CursorReadAsync(string indexName, long cursorId, int? count = null);
 
+    /// <summary>
+    /// Read next results from an existing cursor.
+    /// </summary>
+    /// <param name="result">The result of a previous AggregateAsync or CursorReadAsync call.</param>
+    /// <param name="count">Limit the amount of returned results.</param>
+    /// <returns>A AggregationResult object with the results</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    Task<AggregationResult> CursorReadAsync(AggregationResult result, int? count = null);
+
     /// <summary>
     /// Add terms to a dictionary.
     /// </summary>