diff --git a/.claude/skills/implement-resp-command/SKILL.md b/.claude/skills/implement-resp-command/SKILL.md index 89558763f..04edf7b6e 100644 --- a/.claude/skills/implement-resp-command/SKILL.md +++ b/.claude/skills/implement-resp-command/SKILL.md @@ -28,6 +28,7 @@ Before writing anything, get the command's exact argument order and reply shape 2. **Declare the method on the interfaces** — `src/StackExchange.Redis/Interfaces/IDatabase.cs` *and* `IDatabaseAsync.cs` (or the `.Arrays.cs` / `.VectorSets.cs` partials when relevant). Always provide both sync and async. - **Back-compat:** never add an optional parameter to an existing shipped method (binary break → `MissingMethodException`). Add a new **overload** instead; see `AGENTS.md`. + - **Additive-overload trick (avoids ambiguity):** if the shipped method has an all-optional tail (e.g. `Foo(key, int? count = null, CommandFlags flags = CommandFlags.None)`), a new overload that just appends more optional params (`Foo(key, int? count = null, int? extra = null, CommandFlags flags = ...)`) is **ambiguous** for existing calls like `Foo(key, 5)` — both candidates substitute a default, so neither is "better" (compile error `CS0121`, and the analyzer flags `RS0026`). The fix: **make the existing overload's parameters non-optional (remove the `= ...` defaults) and let the new overload carry all the optionals.** Removing a default is binary-safe (the IL signature is unchanged) and source-safe (old calls simply rebind to the new all-optional overload, which is functionally identical). Do this consistently across the interface *and* every implementor (`RedisDatabase`, `KeyPrefixed`/`KeyPrefixedDatabase`), and update the changed signatures in **`PublicAPI.Shipped.txt`** (optional→required is an edit-in-place there; the new all-optional overload goes in `PublicAPI.Unshipped.txt`). Wrap the new overload in `#pragma warning disable RS0026` since the analyzer still flags competing optional-param overloads even when they're unambiguous. Watch for the "richest" existing overload living in a sibling partial (e.g. `IDatabaseAsync.VectorSets.cs`). - **Implement the new member on every `IDatabase`/`IDatabaseAsync` implementor**, or the build breaks. Chiefly `KeyspaceIsolation/KeyPrefixedDatabase.cs` — and there it must prefix keys via `ToInner(key)`; a stub that forwards without prefixing compiles but **silently breaks keyspace isolation** for the new command. If the command should also be usable in batches/transactions, add it to `IBatch`/`ITransaction` and their implementations (`RedisBatch`/`RedisTransaction`/`KeyPrefixedBatch`) too. 3. **Implement in `RedisDatabase.cs`** (next to the template you picked). The standard shape: diff --git a/src/StackExchange.Redis/Enums/ListMoveCount.cs b/src/StackExchange.Redis/Enums/ListMoveCount.cs new file mode 100644 index 000000000..c72bc626c --- /dev/null +++ b/src/StackExchange.Redis/Enums/ListMoveCount.cs @@ -0,0 +1,32 @@ +using System; + +namespace StackExchange.Redis; + +/// +/// Specifies how the requested count is interpreted when moving multiple list elements via +/// LMOVEM (see ). +/// +public enum ListMoveCount +{ + /// + /// Move up to the requested number of elements (COUNT); fewer are moved when the + /// source list has fewer elements, and a null result is returned when the source is empty. + /// + UpTo, + + /// + /// Move exactly the requested number of elements (EXACTLY); if the source list does + /// not have that many elements, nothing is moved and a null result is returned. + /// + Exactly, +} + +internal static class ListMoveCountExtensions +{ + internal static RedisValue ToLiteral(this ListMoveCount mode) => mode switch + { + ListMoveCount.UpTo => RedisLiterals.COUNT, + ListMoveCount.Exactly => RedisLiterals.EXACTLY, + _ => throw new ArgumentOutOfRangeException(nameof(mode)), + }; +} diff --git a/src/StackExchange.Redis/Enums/ListMoveOrder.cs b/src/StackExchange.Redis/Enums/ListMoveOrder.cs new file mode 100644 index 000000000..b957bce95 --- /dev/null +++ b/src/StackExchange.Redis/Enums/ListMoveOrder.cs @@ -0,0 +1,31 @@ +using System; + +namespace StackExchange.Redis; + +/// +/// Specifies the ordering of the elements moved (and returned) when moving multiple list elements via +/// LMOVEM (see ). +/// +public enum ListMoveOrder +{ + /// + /// Move the elements as a single block (BULK), preserving the source list's relative order. + /// + Bulk, + + /// + /// Move the elements one-by-one (OBO), as if each was popped from the source and pushed to the + /// destination individually. + /// + OneByOne, +} + +internal static class ListMoveOrderExtensions +{ + internal static RedisValue ToLiteral(this ListMoveOrder order) => order switch + { + ListMoveOrder.Bulk => RedisLiterals.BULK, + ListMoveOrder.OneByOne => RedisLiterals.OBO, + _ => throw new ArgumentOutOfRangeException(nameof(order)), + }; +} diff --git a/src/StackExchange.Redis/Enums/RedisCommand.cs b/src/StackExchange.Redis/Enums/RedisCommand.cs index a355c02e6..eeba29413 100644 --- a/src/StackExchange.Redis/Enums/RedisCommand.cs +++ b/src/StackExchange.Redis/Enums/RedisCommand.cs @@ -133,6 +133,7 @@ internal enum RedisCommand LINSERT, LLEN, LMOVE, + LMOVEM, LMPOP, LPOP, LPOS, @@ -191,6 +192,7 @@ internal enum RedisCommand SCARD, SCRIPT, SDIFF, + SDIFFCARD, SDIFFSTORE, SELECT, SENTINEL, @@ -219,6 +221,7 @@ internal enum RedisCommand STRLEN, SUBSCRIBE, SUNION, + SUNIONCARD, SUNIONSTORE, SSCAN, SSUBSCRIBE, @@ -378,6 +381,7 @@ internal static bool IsPrimaryOnly(this RedisCommand command) case RedisCommand.INCREX: case RedisCommand.LINSERT: case RedisCommand.LMOVE: + case RedisCommand.LMOVEM: case RedisCommand.LMPOP: case RedisCommand.LPOP: case RedisCommand.LPUSH: @@ -529,6 +533,7 @@ internal static bool IsPrimaryOnly(this RedisCommand command) case RedisCommand.SCARD: case RedisCommand.SCRIPT: case RedisCommand.SDIFF: + case RedisCommand.SDIFFCARD: case RedisCommand.SELECT: case RedisCommand.SENTINEL: case RedisCommand.SHUTDOWN: @@ -546,6 +551,7 @@ internal static bool IsPrimaryOnly(this RedisCommand command) case RedisCommand.STRLEN: case RedisCommand.SUBSCRIBE: case RedisCommand.SUNION: + case RedisCommand.SUNIONCARD: case RedisCommand.SUNSUBSCRIBE: case RedisCommand.SSCAN: case RedisCommand.SYNC: diff --git a/src/StackExchange.Redis/Enums/SetOperation.cs b/src/StackExchange.Redis/Enums/SetOperation.cs index 9e3449fa7..ddea342e4 100644 --- a/src/StackExchange.Redis/Enums/SetOperation.cs +++ b/src/StackExchange.Redis/Enums/SetOperation.cs @@ -41,6 +41,14 @@ internal static class SetOperationExtensions _ => OutOfRange(operation), }; + internal static RedisCommand ToSetCardinalityCommand(this SetOperation operation) => operation switch + { + SetOperation.Union => RedisCommand.SUNIONCARD, + SetOperation.Intersect => RedisCommand.SINTERCARD, + SetOperation.Difference => RedisCommand.SDIFFCARD, + _ => OutOfRange(operation), + }; + internal static RedisCommand ToSortedSetCommand(this SetOperation operation) => operation switch { SetOperation.Union => RedisCommand.ZUNION, diff --git a/src/StackExchange.Redis/Interfaces/IDatabase.cs b/src/StackExchange.Redis/Interfaces/IDatabase.cs index c0761b9a7..027ac9ae1 100644 --- a/src/StackExchange.Redis/Interfaces/IDatabase.cs +++ b/src/StackExchange.Redis/Interfaces/IDatabase.cs @@ -1370,6 +1370,28 @@ public partial interface IDatabase : IRedis, IDatabaseAsync /// RedisValue ListMove(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, CommandFlags flags = CommandFlags.None); + /// + /// Removes up to (or exactly) elements from the first or last of the list stored at + /// , and pushes them to the first or last of the list stored at , + /// returning the moved elements in destination order. + /// + /// The key of the list to remove from. + /// The key of the list to move to. + /// What side of the list to remove from. + /// What side of the list to move to. + /// The number of elements to move. + /// Whether is an upper bound (COUNT) or an exact requirement (EXACTLY). + /// Whether the elements are moved as a single block (BULK) or one-by-one (OBO). + /// The flags to use for this operation. + /// + /// The elements moved, in destination order; an empty array if there was nothing to move; or + /// when is and the source list does not have elements. + /// + /// +#pragma warning disable RS0026 // competing overloads - disambiguated by the required count parameter + RedisValue[]? ListMove(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode = ListMoveCount.UpTo, ListMoveOrder order = ListMoveOrder.Bulk, CommandFlags flags = CommandFlags.None); +#pragma warning restore RS0026 + /// /// Returns the specified elements of the list stored at key. /// The offsets start and stop are zero-based indexes, with 0 being the first element of the list (the head of the list), 1 being the next element and so on. @@ -1791,9 +1813,37 @@ public partial interface IDatabase : IRedis, IDatabaseAsync /// The number of elements to check (defaults to 0 and means unlimited). /// The flags to use for this operation. /// The cardinality (number of elements) of the set, or 0 if key does not exist. - /// + /// long SetIntersectionLength(RedisKey[] keys, long limit = 0, CommandFlags flags = CommandFlags.None); + /// + /// + /// Returns the set cardinality (number of elements) of the set that would result from combining the sets + /// stored at the given with the specified , without + /// materializing the combined set or returning its members. + /// + /// + /// If the cardinality reaches partway through the computation, the algorithm + /// will exit and yield as the cardinality. + /// + /// + /// The operation used to combine the sets (maps to SUNIONCARD, SINTERCARD or SDIFFCARD). + /// The keys of the sets. + /// The number of elements to check (defaults to 0 and means unlimited). + /// + /// When , appends APPROX to request a HyperLogLog-based estimate rather than an + /// exact count. At the time of writing only (SUNIONCARD) accepts this; + /// it is passed through for any operation and the server will error if it is not supported. + /// + /// The flags to use for this operation. + /// The cardinality (number of elements) of the combined set. + /// + /// , + /// , + /// . + /// + long SetCombineLength(SetOperation operation, RedisKey[] keys, long limit = 0, bool approximate = false, CommandFlags flags = CommandFlags.None); + /// /// Returns the set cardinality (number of elements) of the set stored at key. /// @@ -3040,7 +3090,28 @@ IEnumerable SortedSetScan( /// Equivalent of calling XREAD COUNT num STREAMS key1 key2 id1 id2. /// /// - RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None); + RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags); + + /// + /// Read from multiple streams, applying global caps on the total number of entries and/or the total reply size + /// across all streams (in addition to the per-stream ). + /// + /// Array of streams and the positions from which to begin reading for each stream. + /// The maximum number of messages to return from each stream. + /// The maximum total number of messages to return across all streams (MAXCOUNT); for no cap. + /// The maximum total reply size in bytes across all streams (MAXSIZE); for no cap. + /// The flags to use for this operation. + /// A value of for each stream. + /// + /// + /// Equivalent of calling XREAD COUNT num MAXCOUNT num MAXSIZE num STREAMS key1 key2 id1 id2. + /// MAXCOUNT/MAXSIZE require server 8.10 or above; at least one entry is always returned even if it exceeds . + /// + /// + /// +#pragma warning disable RS0026 // additive overload: countPerStream is required here, so shorter calls still bind to the existing overload + RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None); +#pragma warning restore RS0026 /// /// Read messages from a stream into an associated consumer group. @@ -3133,7 +3204,33 @@ IEnumerable SortedSetScan( /// Equivalent of calling XREADGROUP GROUP groupName consumerName COUNT countPerStream STREAMS stream1 stream2 id1 id2. /// /// - RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, CommandFlags flags = CommandFlags.None); + RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags); + + /// + /// Read from multiple streams into the given consumer group, applying global caps on the total number of entries + /// and/or the total reply size across all streams (in addition to the per-stream ). + /// The consumer group with the given will need to have been created for each stream prior to calling this method. + /// + /// Array of streams and the positions from which to begin reading for each stream. + /// The name of the consumer group. + /// The name of the consumer. + /// The maximum number of messages to return from each stream. + /// When true, the message will not be added to the pending message list. + /// Auto-claim messages that have been idle for at least this long. + /// The maximum total number of messages to return across all streams (MAXCOUNT); for no cap. + /// The maximum total reply size in bytes across all streams (MAXSIZE); for no cap. + /// The flags to use for this operation. + /// A value of for each stream. + /// + /// + /// Equivalent of calling XREADGROUP GROUP groupName consumerName COUNT countPerStream MAXCOUNT num MAXSIZE num STREAMS stream1 stream2 id1 id2. + /// MAXCOUNT/MAXSIZE require server 8.10 or above; at least one entry is always returned even if it exceeds . + /// + /// + /// +#pragma warning disable RS0026 // additive overload: countPerStream/noAck/claimMinIdleTime are required here, so shorter calls still bind to the existing overloads + RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None); +#pragma warning restore RS0026 /// /// Trim the stream to a specified maximum length. diff --git a/src/StackExchange.Redis/Interfaces/IDatabaseAsync.VectorSets.cs b/src/StackExchange.Redis/Interfaces/IDatabaseAsync.VectorSets.cs index 5060390a3..64e8e3878 100644 --- a/src/StackExchange.Redis/Interfaces/IDatabaseAsync.VectorSets.cs +++ b/src/StackExchange.Redis/Interfaces/IDatabaseAsync.VectorSets.cs @@ -96,7 +96,4 @@ System.Collections.Generic.IAsyncEnumerable VectorSetRangeEnumerateA long count = 100, Exclude exclude = Exclude.None, CommandFlags flags = CommandFlags.None); - - /// - Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, CommandFlags flags = CommandFlags.None); } diff --git a/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs b/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs index 40db55cfd..79ca46eac 100644 --- a/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs +++ b/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs @@ -326,6 +326,11 @@ public partial interface IDatabaseAsync : IRedisAsync /// Task ListMoveAsync(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, CommandFlags flags = CommandFlags.None); + /// +#pragma warning disable RS0026 // competing overloads - disambiguated by the required count parameter + Task ListMoveAsync(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode = ListMoveCount.UpTo, ListMoveOrder order = ListMoveOrder.Bulk, CommandFlags flags = CommandFlags.None); +#pragma warning restore RS0026 + /// Task ListRangeAsync(RedisKey key, long start = 0, long stop = -1, CommandFlags flags = CommandFlags.None); @@ -426,6 +431,9 @@ public partial interface IDatabaseAsync : IRedisAsync /// Task SetIntersectionLengthAsync(RedisKey[] keys, long limit = 0, CommandFlags flags = CommandFlags.None); + /// + Task SetCombineLengthAsync(SetOperation operation, RedisKey[] keys, long limit = 0, bool approximate = false, CommandFlags flags = CommandFlags.None); + /// Task SetLengthAsync(RedisKey key, CommandFlags flags = CommandFlags.None); @@ -746,7 +754,12 @@ IAsyncEnumerable SortedSetScanAsync( Task StreamReadAsync(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None); /// - Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None); + Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags); + + /// +#pragma warning disable RS0026 // additive overload: the existing overload's parameters are required, so shorter calls bind here + Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None); +#pragma warning restore RS0026 /// Task StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position, int? count, CommandFlags flags); @@ -763,6 +776,14 @@ IAsyncEnumerable SortedSetScanAsync( /// Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, CommandFlags flags); + /// + Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags); + + /// +#pragma warning disable RS0026 // additive overload: the existing overload's parameters are required, so shorter calls bind there + Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None); +#pragma warning restore RS0026 + /// Task StreamTrimAsync(RedisKey key, int maxLength, bool useApproximateMaxLength, CommandFlags flags); diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs index 952fe0ed3..e09405e4b 100644 --- a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs +++ b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs @@ -321,6 +321,9 @@ public Task ListLengthAsync(RedisKey key, CommandFlags flags = CommandFlag public Task ListMoveAsync(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, CommandFlags flags = CommandFlags.None) => Inner.ListMoveAsync(ToInner(sourceKey), ToInner(destinationKey), sourceSide, destinationSide); + public Task ListMoveAsync(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode = ListMoveCount.UpTo, ListMoveOrder order = ListMoveOrder.Bulk, CommandFlags flags = CommandFlags.None) => + Inner.ListMoveAsync(ToInner(sourceKey), ToInner(destinationKey), sourceSide, destinationSide, count, mode, order, flags); + public Task ListRangeAsync(RedisKey key, long start = 0, long stop = -1, CommandFlags flags = CommandFlags.None) => Inner.ListRangeAsync(ToInner(key), start, stop, flags); @@ -435,6 +438,9 @@ public Task SetContainsAsync(RedisKey key, RedisValue[] values, CommandF public Task SetIntersectionLengthAsync(RedisKey[] keys, long limit = 0, CommandFlags flags = CommandFlags.None) => Inner.SetIntersectionLengthAsync(ToInner(keys), limit, flags); + public Task SetCombineLengthAsync(SetOperation operation, RedisKey[] keys, long limit = 0, bool approximate = false, CommandFlags flags = CommandFlags.None) => + Inner.SetCombineLengthAsync(operation, ToInner(keys), limit, approximate, flags); + public Task SetLengthAsync(RedisKey key, CommandFlags flags = CommandFlags.None) => Inner.SetLengthAsync(ToInner(key), flags); @@ -699,9 +705,12 @@ public Task StreamRangeAsync(RedisKey key, RedisValue? minId = nu public Task StreamReadAsync(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None) => Inner.StreamReadAsync(ToInner(key), position, count, flags); - public Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None) => + public Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags) => Inner.StreamReadAsync(streamPositions, countPerStream, flags); + public Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) => + Inner.StreamReadAsync(streamPositions, countPerStream, maxCount, maxSize, flags); + public Task StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position, int? count, CommandFlags flags) => Inner.StreamReadGroupAsync(ToInner(key), groupName, consumerName, position, count, flags); @@ -717,9 +726,12 @@ public Task StreamReadGroupAsync(StreamPosition[] streamPositions public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, CommandFlags flags = CommandFlags.None) => Inner.StreamReadGroupAsync(streamPositions, groupName, consumerName, countPerStream, noAck, flags); - public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, CommandFlags flags = CommandFlags.None) => + public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags) => Inner.StreamReadGroupAsync(streamPositions, groupName, consumerName, countPerStream, noAck, claimMinIdleTime, flags); + public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) => + Inner.StreamReadGroupAsync(streamPositions, groupName, consumerName, countPerStream, noAck, claimMinIdleTime, maxCount, maxSize, flags); + public Task StreamTrimAsync(RedisKey key, int maxLength, bool useApproximateMaxLength, CommandFlags flags) => Inner.StreamTrimAsync(ToInner(key), maxLength, useApproximateMaxLength, flags); diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs index 2ca7ca384..0d506a180 100644 --- a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs +++ b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs @@ -308,6 +308,9 @@ public long ListLength(RedisKey key, CommandFlags flags = CommandFlags.None) => public RedisValue ListMove(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, CommandFlags flags = CommandFlags.None) => Inner.ListMove(ToInner(sourceKey), ToInner(destinationKey), sourceSide, destinationSide); + public RedisValue[]? ListMove(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode = ListMoveCount.UpTo, ListMoveOrder order = ListMoveOrder.Bulk, CommandFlags flags = CommandFlags.None) => + Inner.ListMove(ToInner(sourceKey), ToInner(destinationKey), sourceSide, destinationSide, count, mode, order, flags); + public RedisValue[] ListRange(RedisKey key, long start = 0, long stop = -1, CommandFlags flags = CommandFlags.None) => Inner.ListRange(ToInner(key), start, stop, flags); @@ -422,6 +425,9 @@ public bool[] SetContains(RedisKey key, RedisValue[] values, CommandFlags flags public long SetIntersectionLength(RedisKey[] keys, long limit = 0, CommandFlags flags = CommandFlags.None) => Inner.SetIntersectionLength(ToInner(keys), limit, flags); + public long SetCombineLength(SetOperation operation, RedisKey[] keys, long limit = 0, bool approximate = false, CommandFlags flags = CommandFlags.None) => + Inner.SetCombineLength(operation, ToInner(keys), limit, approximate, flags); + public long SetLength(RedisKey key, CommandFlags flags = CommandFlags.None) => Inner.SetLength(ToInner(key), flags); @@ -681,9 +687,12 @@ public StreamEntry[] StreamRange(RedisKey key, RedisValue? minId = null, RedisVa public StreamEntry[] StreamRead(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None) => Inner.StreamRead(ToInner(key), position, count, flags); - public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None) => + public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags) => Inner.StreamRead(streamPositions, countPerStream, flags); + public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) => + Inner.StreamRead(streamPositions, countPerStream, maxCount, maxSize, flags); + public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position, int? count, CommandFlags flags) => Inner.StreamReadGroup(ToInner(key), groupName, consumerName, position, count, flags); @@ -699,9 +708,12 @@ public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValu public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, CommandFlags flags = CommandFlags.None) => Inner.StreamReadGroup(streamPositions, groupName, consumerName, countPerStream, noAck, flags); - public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, CommandFlags flags = CommandFlags.None) => + public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags) => Inner.StreamReadGroup(streamPositions, groupName, consumerName, countPerStream, noAck, claimMinIdleTime, flags); + public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) => + Inner.StreamReadGroup(streamPositions, groupName, consumerName, countPerStream, noAck, claimMinIdleTime, maxCount, maxSize, flags); + public long StreamTrim(RedisKey key, int maxLength, bool useApproximateMaxLength, CommandFlags flags) => Inner.StreamTrim(ToInner(key), maxLength, useApproximateMaxLength, flags); diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt index 7e8381c6e..686586690 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt @@ -760,7 +760,7 @@ StackExchange.Redis.IDatabase.StreamPendingMessages(StackExchange.Redis.RedisKey StackExchange.Redis.IDatabase.StreamPendingMessages(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, int count, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? minId = null, StackExchange.Redis.RedisValue? maxId = null, long? minIdleTimeInMs = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.StreamPendingMessageInfo[]! StackExchange.Redis.IDatabase.StreamRange(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue? minId = null, StackExchange.Redis.RedisValue? maxId = null, int? count = null, StackExchange.Redis.Order messageOrder = StackExchange.Redis.Order.Ascending, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.StreamEntry[]! StackExchange.Redis.IDatabase.StreamRead(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue position, int? count = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.StreamEntry[]! -StackExchange.Redis.IDatabase.StreamRead(StackExchange.Redis.StreamPosition[]! streamPositions, int? countPerStream = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisStream[]! +StackExchange.Redis.IDatabase.StreamRead(StackExchange.Redis.StreamPosition[]! streamPositions, int? countPerStream, StackExchange.Redis.CommandFlags flags) -> StackExchange.Redis.RedisStream[]! StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? position, int? count, bool noAck, StackExchange.Redis.CommandFlags flags) -> StackExchange.Redis.StreamEntry[]! StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? position = null, int? count = null, bool noAck = false, System.TimeSpan? claimMinIdleTime = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.StreamEntry[]! StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? position, int? count, StackExchange.Redis.CommandFlags flags) -> StackExchange.Redis.StreamEntry[]! @@ -1006,7 +1006,7 @@ StackExchange.Redis.IDatabaseAsync.StreamPendingMessagesAsync(StackExchange.Redi StackExchange.Redis.IDatabaseAsync.StreamPendingMessagesAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, int count, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? minId = null, StackExchange.Redis.RedisValue? maxId = null, long? minIdleTimeInMs = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! StackExchange.Redis.IDatabaseAsync.StreamRangeAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue? minId = null, StackExchange.Redis.RedisValue? maxId = null, int? count = null, StackExchange.Redis.Order messageOrder = StackExchange.Redis.Order.Ascending, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! StackExchange.Redis.IDatabaseAsync.StreamReadAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue position, int? count = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! -StackExchange.Redis.IDatabaseAsync.StreamReadAsync(StackExchange.Redis.StreamPosition[]! streamPositions, int? countPerStream = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.IDatabaseAsync.StreamReadAsync(StackExchange.Redis.StreamPosition[]! streamPositions, int? countPerStream, StackExchange.Redis.CommandFlags flags) -> System.Threading.Tasks.Task! StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? position, int? count, bool noAck, StackExchange.Redis.CommandFlags flags) -> System.Threading.Tasks.Task! StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? position = null, int? count = null, bool noAck = false, System.TimeSpan? claimMinIdleTime = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.RedisValue? position, int? count, StackExchange.Redis.CommandFlags flags) -> System.Threading.Tasks.Task! @@ -2070,9 +2070,9 @@ static StackExchange.Redis.VectorSetAddRequest.Member(StackExchange.Redis.RedisV static StackExchange.Redis.VectorSetSimilaritySearchRequest.ByMember(StackExchange.Redis.RedisValue member) -> StackExchange.Redis.VectorSetSimilaritySearchRequest! static StackExchange.Redis.VectorSetSimilaritySearchRequest.ByVector(System.ReadOnlyMemory vector) -> StackExchange.Redis.VectorSetSimilaritySearchRequest! StackExchange.Redis.RedisChannel.WithKeyRouting() -> StackExchange.Redis.RedisChannel -StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream = null, bool noAck = false, System.TimeSpan? claimMinIdleTime = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisStream[]! +StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream, bool noAck, System.TimeSpan? claimMinIdleTime, StackExchange.Redis.CommandFlags flags) -> StackExchange.Redis.RedisStream[]! StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream, bool noAck, StackExchange.Redis.CommandFlags flags) -> StackExchange.Redis.RedisStream[]! -StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream = null, bool noAck = false, System.TimeSpan? claimMinIdleTime = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream, bool noAck, System.TimeSpan? claimMinIdleTime, StackExchange.Redis.CommandFlags flags) -> System.Threading.Tasks.Task! StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream, bool noAck, StackExchange.Redis.CommandFlags flags) -> System.Threading.Tasks.Task! StackExchange.Redis.StreamEntry.DeliveryCount.get -> int StackExchange.Redis.StreamEntry.IdleTime.get -> System.TimeSpan? diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 7dc5c5811..60440ce7b 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1 +1,15 @@ #nullable enable +StackExchange.Redis.IDatabase.ListMove(StackExchange.Redis.RedisKey sourceKey, StackExchange.Redis.RedisKey destinationKey, StackExchange.Redis.ListSide sourceSide, StackExchange.Redis.ListSide destinationSide, long count, StackExchange.Redis.ListMoveCount mode = StackExchange.Redis.ListMoveCount.UpTo, StackExchange.Redis.ListMoveOrder order = StackExchange.Redis.ListMoveOrder.Bulk, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisValue[]? +StackExchange.Redis.IDatabase.SetCombineLength(StackExchange.Redis.SetOperation operation, StackExchange.Redis.RedisKey[]! keys, long limit = 0, bool approximate = false, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long +StackExchange.Redis.IDatabase.StreamRead(StackExchange.Redis.StreamPosition[]! streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisStream[]! +StackExchange.Redis.IDatabase.StreamReadGroup(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream = null, bool noAck = false, System.TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisStream[]! +StackExchange.Redis.IDatabaseAsync.ListMoveAsync(StackExchange.Redis.RedisKey sourceKey, StackExchange.Redis.RedisKey destinationKey, StackExchange.Redis.ListSide sourceSide, StackExchange.Redis.ListSide destinationSide, long count, StackExchange.Redis.ListMoveCount mode = StackExchange.Redis.ListMoveCount.UpTo, StackExchange.Redis.ListMoveOrder order = StackExchange.Redis.ListMoveOrder.Bulk, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.IDatabaseAsync.SetCombineLengthAsync(StackExchange.Redis.SetOperation operation, StackExchange.Redis.RedisKey[]! keys, long limit = 0, bool approximate = false, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.IDatabaseAsync.StreamReadAsync(StackExchange.Redis.StreamPosition[]! streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.IDatabaseAsync.StreamReadGroupAsync(StackExchange.Redis.StreamPosition[]! streamPositions, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, int? countPerStream = null, bool noAck = false, System.TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.ListMoveCount +StackExchange.Redis.ListMoveCount.Exactly = 1 -> StackExchange.Redis.ListMoveCount +StackExchange.Redis.ListMoveCount.UpTo = 0 -> StackExchange.Redis.ListMoveCount +StackExchange.Redis.ListMoveOrder +StackExchange.Redis.ListMoveOrder.Bulk = 0 -> StackExchange.Redis.ListMoveOrder +StackExchange.Redis.ListMoveOrder.OneByOne = 1 -> StackExchange.Redis.ListMoveOrder diff --git a/src/StackExchange.Redis/RedisDatabase.cs b/src/StackExchange.Redis/RedisDatabase.cs index 21518eaa7..fddbb93a9 100644 --- a/src/StackExchange.Redis/RedisDatabase.cs +++ b/src/StackExchange.Redis/RedisDatabase.cs @@ -1630,6 +1630,21 @@ public Task ListMoveAsync(RedisKey sourceKey, RedisKey destinationKe return ExecuteAsync(msg, ResultProcessor.RedisValue); } + public RedisValue[]? ListMove(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode = ListMoveCount.UpTo, ListMoveOrder order = ListMoveOrder.Bulk, CommandFlags flags = CommandFlags.None) + { + var msg = GetListMoveMultipleMessage(sourceKey, destinationKey, sourceSide, destinationSide, count, mode, order, flags); + return ExecuteSync(msg, ResultProcessor.NullableRedisValueArray); + } + + public Task ListMoveAsync(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode = ListMoveCount.UpTo, ListMoveOrder order = ListMoveOrder.Bulk, CommandFlags flags = CommandFlags.None) + { + var msg = GetListMoveMultipleMessage(sourceKey, destinationKey, sourceSide, destinationSide, count, mode, order, flags); + return ExecuteAsync(msg, ResultProcessor.NullableRedisValueArray); + } + + private Message GetListMoveMultipleMessage(RedisKey sourceKey, RedisKey destinationKey, ListSide sourceSide, ListSide destinationSide, long count, ListMoveCount mode, ListMoveOrder order, CommandFlags flags) => + Message.Create(Database, flags, RedisCommand.LMOVEM, sourceKey, destinationKey, sourceSide.ToLiteral(), destinationSide.ToLiteral(), mode.ToLiteral(), count, order.ToLiteral()); + public RedisValue[] ListRange(RedisKey key, long start = 0, long stop = -1, CommandFlags flags = CommandFlags.None) { var msg = Message.Create(Database, flags, RedisCommand.LRANGE, key, start, stop); @@ -2142,6 +2157,18 @@ public Task SetIntersectionLengthAsync(RedisKey[] keys, long limit = 0, Co return ExecuteAsync(msg, ResultProcessor.Int64); } + public long SetCombineLength(SetOperation operation, RedisKey[] keys, long limit = 0, bool approximate = false, CommandFlags flags = CommandFlags.None) + { + var msg = GetSetCombineLengthMessage(operation, keys, limit, approximate, flags); + return ExecuteSync(msg, ResultProcessor.Int64); + } + + public Task SetCombineLengthAsync(SetOperation operation, RedisKey[] keys, long limit = 0, bool approximate = false, CommandFlags flags = CommandFlags.None) + { + var msg = GetSetCombineLengthMessage(operation, keys, limit, approximate, flags); + return ExecuteAsync(msg, ResultProcessor.Int64); + } + public long SetLength(RedisKey key, CommandFlags flags = CommandFlags.None) { var msg = Message.Create(Database, flags, RedisCommand.SCARD, key); @@ -3428,18 +3455,32 @@ public Task StreamReadAsync(RedisKey key, RedisValue position, in return ExecuteAsync(msg, ResultProcessor.SingleStreamWithNameSkip, defaultValue: Array.Empty()); } - public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None) + public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags) { var msg = GetMultiStreamReadMessage(streamPositions, countPerStream, flags); return ExecuteSync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); } - public Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None) + public Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags) { var msg = GetMultiStreamReadMessage(streamPositions, countPerStream, flags); return ExecuteAsync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); } +#pragma warning disable RS0026 // additive overload: the existing overload's parameters are required, so shorter calls bind here + public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) + { + var msg = GetMultiStreamReadMessage(streamPositions, countPerStream, flags, maxCount, maxSize); + return ExecuteSync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); + } + + public Task StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) + { + var msg = GetMultiStreamReadMessage(streamPositions, countPerStream, flags, maxCount, maxSize); + return ExecuteAsync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); + } +#pragma warning restore RS0026 + public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position, int? count, CommandFlags flags) => StreamReadGroup( key, @@ -3536,7 +3577,7 @@ public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValu null, flags); - public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, CommandFlags flags = CommandFlags.None) + public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags) { var msg = GetMultiStreamReadGroupMessage( streamPositions, @@ -3570,7 +3611,7 @@ public Task StreamReadGroupAsync(StreamPosition[] streamPositions null, flags); - public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, CommandFlags flags = CommandFlags.None) + public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags) { var msg = GetMultiStreamReadGroupMessage( streamPositions, @@ -3584,6 +3625,40 @@ public Task StreamReadGroupAsync(StreamPosition[] streamPositions return ExecuteAsync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); } +#pragma warning disable RS0026 // additive overload: the existing overload's parameters are required, so shorter calls bind here + public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) + { + var msg = GetMultiStreamReadGroupMessage( + streamPositions, + groupName, + consumerName, + countPerStream, + noAck, + claimMinIdleTime, + flags, + maxCount, + maxSize); + + return ExecuteSync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); + } + + public Task StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, bool noAck = false, TimeSpan? claimMinIdleTime = null, int? maxCount = null, int? maxSize = null, CommandFlags flags = CommandFlags.None) + { + var msg = GetMultiStreamReadGroupMessage( + streamPositions, + groupName, + consumerName, + countPerStream, + noAck, + claimMinIdleTime, + flags, + maxCount, + maxSize); + + return ExecuteAsync(msg, ResultProcessor.MultiStream, defaultValue: Array.Empty()); + } +#pragma warning restore RS0026 + public long StreamTrim(RedisKey key, int maxLength, bool useApproximateMaxLength, CommandFlags flags) => StreamTrim(key, maxLength, useApproximateMaxLength, null, StreamTrimMode.KeepReferences, flags); @@ -4216,7 +4291,7 @@ internal static RedisValue GetLexRange(RedisValue value, Exclude exclude, bool i return result; } - private Message GetMultiStreamReadGroupMessage(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags) => + private Message GetMultiStreamReadGroupMessage(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, CommandFlags flags, int? maxCount = null, int? maxSize = null) => new MultiStreamReadGroupCommandMessage( Database, flags, @@ -4225,19 +4300,23 @@ private Message GetMultiStreamReadGroupMessage(StreamPosition[] streamPositions, consumerName, countPerStream, noAck, - claimMinIdleTime); + claimMinIdleTime, + maxCount, + maxSize); - private sealed class MultiStreamReadGroupCommandMessage : Message // XREADGROUP with multiple stream. Example: XREADGROUP GROUP groupName consumerName COUNT countPerStream STREAMS stream1 stream2 id1 id2 + internal sealed class MultiStreamReadGroupCommandMessage : Message // XREADGROUP with multiple stream. Example: XREADGROUP GROUP groupName consumerName COUNT countPerStream STREAMS stream1 stream2 id1 id2 { private readonly StreamPosition[] streamPositions; private readonly RedisValue groupName; private readonly RedisValue consumerName; private readonly int? countPerStream; + private readonly int? maxCount; + private readonly int? maxSize; private readonly bool noAck; private readonly int argCount; private readonly TimeSpan? claimMinIdleTime; - public MultiStreamReadGroupCommandMessage(int db, CommandFlags flags, StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime) + public MultiStreamReadGroupCommandMessage(int db, CommandFlags flags, StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream, bool noAck, TimeSpan? claimMinIdleTime, int? maxCount = null, int? maxSize = null) : base(db, flags, RedisCommand.XREADGROUP) { if (streamPositions == null) throw new ArgumentNullException(nameof(streamPositions)); @@ -4252,6 +4331,16 @@ public MultiStreamReadGroupCommandMessage(int db, CommandFlags flags, StreamPosi throw new ArgumentOutOfRangeException(nameof(countPerStream), "countPerStream must be greater than 0."); } + if (maxCount.HasValue && maxCount <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxCount), "maxCount must be greater than 0."); + } + + if (maxSize.HasValue && maxSize <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxSize), "maxSize must be greater than 0."); + } + groupName.AssertNotNull(); consumerName.AssertNotNull(); @@ -4259,12 +4348,16 @@ public MultiStreamReadGroupCommandMessage(int db, CommandFlags flags, StreamPosi this.groupName = groupName; this.consumerName = consumerName; this.countPerStream = countPerStream; + this.maxCount = maxCount; + this.maxSize = maxSize; this.noAck = noAck; this.claimMinIdleTime = claimMinIdleTime; argCount = 4 // Room for GROUP groupName consumerName & STREAMS + (streamPositions.Length * 2) // Enough room for the stream keys and associated IDs. + (countPerStream.HasValue ? 2 : 0) // Room for "COUNT num" or 0 if countPerStream is null. + + (maxCount.HasValue ? 2 : 0) // Room for "MAXCOUNT num". + + (maxSize.HasValue ? 2 : 0) // Room for "MAXSIZE num". + (noAck ? 1 : 0) // Allow for the NOACK subcommand. + (claimMinIdleTime.HasValue ? 2 : 0); // Allow for the CLAIM {minIdleTime} subcommand. } @@ -4292,6 +4385,18 @@ protected override void WriteImpl(in MessageWriter writer) writer.WriteBulkString(countPerStream.Value); } + if (maxCount.HasValue) + { + writer.WriteRaw("$8\r\nMAXCOUNT\r\n"u8); + writer.WriteBulkString(maxCount.Value); + } + + if (maxSize.HasValue) + { + writer.WriteRaw("$7\r\nMAXSIZE\r\n"u8); + writer.WriteBulkString(maxSize.Value); + } + if (noAck) { writer.WriteRaw("$5\r\nNOACK\r\n"u8); @@ -4317,16 +4422,18 @@ protected override void WriteImpl(in MessageWriter writer) public override int ArgCount => argCount; } - private Message GetMultiStreamReadMessage(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags) => - new MultiStreamReadCommandMessage(Database, flags, streamPositions, countPerStream); + private Message GetMultiStreamReadMessage(StreamPosition[] streamPositions, int? countPerStream, CommandFlags flags, int? maxCount = null, int? maxSize = null) => + new MultiStreamReadCommandMessage(Database, flags, streamPositions, countPerStream, maxCount, maxSize); - private sealed class MultiStreamReadCommandMessage : Message // XREAD with multiple stream. Example: XREAD COUNT 2 STREAMS mystream writers 0-0 0-0 + internal sealed class MultiStreamReadCommandMessage : Message // XREAD with multiple stream. Example: XREAD COUNT 2 STREAMS mystream writers 0-0 0-0 { private readonly StreamPosition[] streamPositions; private readonly int? countPerStream; + private readonly int? maxCount; + private readonly int? maxSize; private readonly int argCount; - public MultiStreamReadCommandMessage(int db, CommandFlags flags, StreamPosition[] streamPositions, int? countPerStream) + public MultiStreamReadCommandMessage(int db, CommandFlags flags, StreamPosition[] streamPositions, int? countPerStream, int? maxCount = null, int? maxSize = null) : base(db, flags, RedisCommand.XREAD) { if (streamPositions == null) throw new ArgumentNullException(nameof(streamPositions)); @@ -4341,11 +4448,25 @@ public MultiStreamReadCommandMessage(int db, CommandFlags flags, StreamPosition[ throw new ArgumentOutOfRangeException(nameof(countPerStream), "countPerStream must be greater than 0."); } + if (maxCount.HasValue && maxCount <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxCount), "maxCount must be greater than 0."); + } + + if (maxSize.HasValue && maxSize <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxSize), "maxSize must be greater than 0."); + } + this.streamPositions = streamPositions; this.countPerStream = countPerStream; + this.maxCount = maxCount; + this.maxSize = maxSize; argCount = 1 // Streams keyword. + (countPerStream.HasValue ? 2 : 0) // Room for "COUNT num" or 0 if countPerStream is null. + + (maxCount.HasValue ? 2 : 0) // Room for "MAXCOUNT num". + + (maxSize.HasValue ? 2 : 0) // Room for "MAXSIZE num". + (streamPositions.Length * 2); // Room for the stream names and the ID after which to begin reading. } @@ -4369,6 +4490,18 @@ protected override void WriteImpl(in MessageWriter writer) writer.WriteBulkString(countPerStream.Value); } + if (maxCount.HasValue) + { + writer.WriteRaw("$8\r\nMAXCOUNT\r\n"u8); + writer.WriteBulkString(maxCount.Value); + } + + if (maxSize.HasValue) + { + writer.WriteRaw("$7\r\nMAXSIZE\r\n"u8); + writer.WriteBulkString(maxSize.Value); + } + writer.WriteRaw("$7\r\nSTREAMS\r\n"u8); for (int i = 0; i < streamPositions.Length; i++) { @@ -4403,7 +4536,13 @@ private Message GetRestoreMessage(RedisKey key, byte[] value, TimeSpan? expiry, } private Message GetSetCardinalityMessage(RedisCommand command, RedisKey[] keys, long limit, CommandFlags flags) - => new SetOperationCardinalityMessage(Database, flags, command, keys, limit); + => new SetOperationCardinalityMessage(Database, flags, command, keys, limit, approximate: false); + + private Message GetSetCombineLengthMessage(SetOperation operation, RedisKey[] keys, long limit, bool approximate, CommandFlags flags) + // Deliberately do NOT gate `approximate` here: today only SUNIONCARD accepts APPROX (SINTERCARD/SDIFFCARD + // will error), but if later server versions extend it we don't want a stale client-side check blocking it. + // Let the server decide and surface any error. + => new SetOperationCardinalityMessage(Database, flags, operation.ToSetCardinalityCommand(), keys, limit, approximate); private Message GetSortedSetAddMessage(RedisKey key, RedisValue member, double score, SortedSetWhen when, bool change, CommandFlags flags) => new SingleSortedSetAddMessage(Database, flags, key, member, score, when, change, increment: false); @@ -4593,7 +4732,7 @@ private Message GetSortedSetLengthMessage(RedisKey key, double min, double max, } private Message GetSortedSetIntersectionLengthMessage(RedisKey[] keys, long limit, CommandFlags flags) - => new SetOperationCardinalityMessage(Database, flags, RedisCommand.ZINTERCARD, keys, limit); + => new SetOperationCardinalityMessage(Database, flags, RedisCommand.ZINTERCARD, keys, limit, approximate: false); private Message GetSortedSetRangeByScoreMessage(RedisKey key, double start, double stop, Exclude exclude, Order order, long skip, long take, CommandFlags flags, bool withScores) { diff --git a/src/StackExchange.Redis/RedisFeatures.cs b/src/StackExchange.Redis/RedisFeatures.cs index 1a40cd427..a2d77e959 100644 --- a/src/StackExchange.Redis/RedisFeatures.cs +++ b/src/StackExchange.Redis/RedisFeatures.cs @@ -50,7 +50,8 @@ namespace StackExchange.Redis v8_2_0_rc1 = new Version(8, 1, 240), // 8.2 RC1 is version 8.1.240 v8_4_0_rc1 = new Version(8, 3, 224), // 8.4 RC1 is version 8.3.224 v8_6_0 = new Version(8, 6, 0), - v8_8_0 = new Version(8, 7, 225); // 8.8 is version 8.7.225 + v8_8_0 = new Version(8, 7, 225), // 8.8 is version 8.7.225 + v8_10_0_rc2 = new Version(8, 9, 241); // 8.10 RC2 is version 8.9.241 #pragma warning restore SA1310 // Field names should not contain underscore #pragma warning restore SA1311 // Static readonly fields should begin with upper-case letter diff --git a/src/StackExchange.Redis/RedisLiterals.cs b/src/StackExchange.Redis/RedisLiterals.cs index 9709eb02e..5d1f20b91 100644 --- a/src/StackExchange.Redis/RedisLiterals.cs +++ b/src/StackExchange.Redis/RedisLiterals.cs @@ -20,6 +20,7 @@ public static readonly RedisValue ASC = RedisValue.FromRaw("ASC"u8), BEFORE = RedisValue.FromRaw("BEFORE"u8), BIT = RedisValue.FromRaw("BIT"u8), + BULK = RedisValue.FromRaw("BULK"u8), BY = RedisValue.FromRaw("BY"u8), BYLEX = RedisValue.FromRaw("BYLEX"u8), BYSCORE = RedisValue.FromRaw("BYSCORE"u8), @@ -35,6 +36,7 @@ public static readonly RedisValue DOCTOR = RedisValue.FromRaw("DOCTOR"u8), ENCODING = RedisValue.FromRaw("ENCODING"u8), EX = RedisValue.FromRaw("EX"u8), + EXACTLY = RedisValue.FromRaw("EXACTLY"u8), EXAT = RedisValue.FromRaw("EXAT"u8), EXISTS = RedisValue.FromRaw("EXISTS"u8), FIELDS = RedisValue.FromRaw("FIELDS"u8), @@ -87,6 +89,7 @@ public static readonly RedisValue NUMSUB = RedisValue.FromRaw("NUMSUB"u8), NX = RedisValue.FromRaw("NX"u8), OBJECT = RedisValue.FromRaw("OBJECT"u8), + OBO = RedisValue.FromRaw("OBO"u8), ONE = RedisValue.FromRaw("ONE"u8), OR = RedisValue.FromRaw("OR"u8), PATTERN = RedisValue.FromRaw("PATTERN"u8), diff --git a/src/StackExchange.Redis/ResultProcessor.ListMove.cs b/src/StackExchange.Redis/ResultProcessor.ListMove.cs new file mode 100644 index 000000000..2d342aec0 --- /dev/null +++ b/src/StackExchange.Redis/ResultProcessor.ListMove.cs @@ -0,0 +1,34 @@ +using RESPite.Messages; + +// ReSharper disable once CheckNamespace +namespace StackExchange.Redis; + +internal abstract partial class ResultProcessor +{ + /// + /// Parses the reply of LMOVEM: an array of moved elements, or when + /// EXACTLY could not be satisfied. A null reply is preserved as , distinct + /// from an empty array (nothing moved under COUNT). + /// + public static readonly ResultProcessor + NullableRedisValueArray = new NullableRedisValueArrayProcessor(); + + private sealed class NullableRedisValueArrayProcessor : ResultProcessor + { + protected override bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader) + { + if (reader.IsNull) + { + SetResult(message, null); + return true; + } + if (reader.IsAggregate) + { + var arr = reader.ReadPastRedisValues() ?? []; + SetResult(message, arr); + return true; + } + return false; // scalar / other => unexpected + } + } +} diff --git a/src/StackExchange.Redis/SetOperationCardinalityMessage.cs b/src/StackExchange.Redis/SetOperationCardinalityMessage.cs index 511923c40..5fb699333 100644 --- a/src/StackExchange.Redis/SetOperationCardinalityMessage.cs +++ b/src/StackExchange.Redis/SetOperationCardinalityMessage.cs @@ -5,11 +5,12 @@ internal sealed class SetOperationCardinalityMessage( CommandFlags flags, RedisCommand command, RedisKey[] keys, - long limit) : Message(db, flags, command) + long limit, + bool approximate) : Message(db, flags, command) { private readonly RedisKey[] _keys = keys.AssertAllNonNull(); - public override int ArgCount => 1 + _keys.Length + (limit > 0 ? 2 : 0); + public override int ArgCount => 1 + _keys.Length + (approximate ? 1 : 0) + (limit > 0 ? 2 : 0); public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) => serverSelectionStrategy.HashSlot(_keys); @@ -22,6 +23,11 @@ protected override void WriteImpl(in MessageWriter writer) writer.Write(_keys[i]); } + if (approximate) + { + writer.WriteRaw("$6\r\nAPPROX\r\n"u8); + } + if (limit > 0) { writer.WriteRaw("$5\r\nLIMIT\r\n"u8); diff --git a/tests/RedisConfigs/.docker/Redis/Dockerfile b/tests/RedisConfigs/.docker/Redis/Dockerfile index f5e3f196a..0a00dac88 100644 --- a/tests/RedisConfigs/.docker/Redis/Dockerfile +++ b/tests/RedisConfigs/.docker/Redis/Dockerfile @@ -1,4 +1,4 @@ -ARG CLIENT_LIBS_TEST_IMAGE=redislabs/client-libs-test:8.8.0 +ARG CLIENT_LIBS_TEST_IMAGE=redislabs/client-libs-test:8.10-rc2 FROM ${CLIENT_LIBS_TEST_IMAGE} COPY --from=configs ./Basic /data/Basic/ diff --git a/tests/RedisConfigs/docker-compose.yml b/tests/RedisConfigs/docker-compose.yml index ce5475aaf..62d89150d 100644 --- a/tests/RedisConfigs/docker-compose.yml +++ b/tests/RedisConfigs/docker-compose.yml @@ -5,7 +5,7 @@ services: build: context: .docker/Redis args: - CLIENT_LIBS_TEST_IMAGE: ${CLIENT_LIBS_TEST_IMAGE:-redislabs/client-libs-test:8.8.0} + CLIENT_LIBS_TEST_IMAGE: ${CLIENT_LIBS_TEST_IMAGE:-redislabs/client-libs-test:8.10-rc2} additional_contexts: configs: . platform: linux diff --git a/tests/StackExchange.Redis.Tests/KeyPrefixedDatabaseTests.cs b/tests/StackExchange.Redis.Tests/KeyPrefixedDatabaseTests.cs index ac8d6b05d..7ab5fa85f 100644 --- a/tests/StackExchange.Redis.Tests/KeyPrefixedDatabaseTests.cs +++ b/tests/StackExchange.Redis.Tests/KeyPrefixedDatabaseTests.cs @@ -679,6 +679,13 @@ public void SetIntersectionLength() mock.Received().SetIntersectionLength(IsKeys(["prefix:key1", "prefix:key2"]), 0, CommandFlags.None); } + [Fact] + public void SetCombineLength() + { + prefixed.SetCombineLength(SetOperation.Union, ["key1", "key2"]); + mock.Received().SetCombineLength(SetOperation.Union, IsKeys(["prefix:key1", "prefix:key2"]), 0, false, CommandFlags.None); + } + [Fact] public void SetLength() { diff --git a/tests/StackExchange.Redis.Tests/KeyPrefixedTests.cs b/tests/StackExchange.Redis.Tests/KeyPrefixedTests.cs index 58d333ec9..8ab933deb 100644 --- a/tests/StackExchange.Redis.Tests/KeyPrefixedTests.cs +++ b/tests/StackExchange.Redis.Tests/KeyPrefixedTests.cs @@ -632,6 +632,13 @@ public async Task SetIntersectionLengthAsync() await mock.Received().SetIntersectionLengthAsync(IsKeys("prefix:key1", "prefix:key2"), 0, CommandFlags.None); } + [Fact] + public async Task SetCombineLengthAsync() + { + await prefixed.SetCombineLengthAsync(SetOperation.Union, ["key1", "key2"]); + await mock.Received().SetCombineLengthAsync(SetOperation.Union, IsKeys("prefix:key1", "prefix:key2"), 0, false, CommandFlags.None); + } + [Fact] public async Task SetLengthAsync() { diff --git a/tests/StackExchange.Redis.Tests/ListTests.cs b/tests/StackExchange.Redis.Tests/ListTests.cs index cd0f2e0a3..16171d542 100644 --- a/tests/StackExchange.Redis.Tests/ListTests.cs +++ b/tests/StackExchange.Redis.Tests/ListTests.cs @@ -327,6 +327,83 @@ public async Task ListMoveKeyDoesNotExist() Assert.True(rangeResult1.IsNull); } + [Fact] + public async Task ListMoveMultiple_UpTo() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + RedisKey src = Me(); + RedisKey dest = Me() + "dest"; + db.KeyDelete([src, dest], CommandFlags.FireAndForget); + + await db.ListRightPushAsync(src, ["a", "b", "c", "d"]); + + // move up-to 2 from the head of src onto the tail of dest + var moved = await db.ListMoveAsync(src, dest, ListSide.Left, ListSide.Right, 2); + Assert.NotNull(moved); + Assert.Equal(["a", "b"], moved.Select(x => x.ToString())); + Assert.Equal(["c", "d"], db.ListRange(src).Select(x => x.ToString())); + Assert.Equal(["a", "b"], db.ListRange(dest).Select(x => x.ToString())); + + // COUNT tolerates asking for more than exist: moves what's left (here, the final 2) + var rest = db.ListMove(src, dest, ListSide.Left, ListSide.Right, 100); + Assert.NotNull(rest); + Assert.Equal(["c", "d"], rest.Select(x => x.ToString())); + Assert.Equal(0, db.ListLength(src)); + + // COUNT against an empty source moves nothing and returns null (as LMOVE does) + var none = db.ListMove(src, dest, ListSide.Left, ListSide.Right, 5); + Assert.Null(none); + } + + [Fact] + public async Task ListMoveMultiple_Ordering() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + RedisKey src = Me(); + RedisKey dest = Me() + "dest"; + + // BULK preserves the source's relative order... + db.KeyDelete([src, dest], CommandFlags.FireAndForget); + await db.ListRightPushAsync(src, ["a", "b", "c", "d"]); + var bulk = db.ListMove(src, dest, ListSide.Left, ListSide.Left, 2, ListMoveCount.UpTo, ListMoveOrder.Bulk); + Assert.Equal(["a", "b"], bulk!.Select(x => x.ToString())); + + // ...whereas OBO moves them one-by-one, reversing them when pushed onto the head. + db.KeyDelete([src, dest], CommandFlags.FireAndForget); + await db.ListRightPushAsync(src, ["a", "b", "c", "d"]); + var obo = db.ListMove(src, dest, ListSide.Left, ListSide.Left, 2, ListMoveCount.UpTo, ListMoveOrder.OneByOne); + Assert.Equal(["b", "a"], obo!.Select(x => x.ToString())); + } + + [Fact] + public async Task ListMoveMultiple_Exactly() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + RedisKey src = Me(); + RedisKey dest = Me() + "dest"; + db.KeyDelete([src, dest], CommandFlags.FireAndForget); + + await db.ListRightPushAsync(src, ["a", "b"]); + + // EXACTLY is all-or-nothing: too few elements => null, and the source is left untouched + var unsatisfied = db.ListMove(src, dest, ListSide.Left, ListSide.Right, 3, ListMoveCount.Exactly); + Assert.Null(unsatisfied); + Assert.Equal(["a", "b"], db.ListRange(src).Select(x => x.ToString())); + Assert.Equal(0, db.ListLength(dest)); + + // exactly the right number => all move + var satisfied = db.ListMove(src, dest, ListSide.Left, ListSide.Right, 2, ListMoveCount.Exactly); + Assert.NotNull(satisfied); + Assert.Equal(["a", "b"], satisfied.Select(x => x.ToString())); + Assert.Equal(0, db.ListLength(src)); + } + [Fact] public async Task ListPositionHappyPath() { diff --git a/tests/StackExchange.Redis.Tests/ResultProcessorUnitTests/ListMoveMultiple.cs b/tests/StackExchange.Redis.Tests/ResultProcessorUnitTests/ListMoveMultiple.cs new file mode 100644 index 000000000..d5a884a68 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ResultProcessorUnitTests/ListMoveMultiple.cs @@ -0,0 +1,69 @@ +using Xunit; + +namespace StackExchange.Redis.Tests.ResultProcessorUnitTests; + +public class ListMoveMultiple(ITestOutputHelper log) : ResultProcessorUnitTest(log) +{ + [Fact] + public void MovedElements_Success() + { + // LMOVEM src dst LEFT RIGHT COUNT 2 BULK => ["a", "b"] + var resp = "*2\r\n$1\r\na\r\n$1\r\nb\r\n"; + + var result = Execute(resp, ResultProcessor.NullableRedisValueArray); + + Assert.NotNull(result); + Assert.Equal("a,b", Join(result)); + } + + [Fact] + public void EmptyArray_IsEmptyNotNull() + { + // an empty array must stay an empty array, distinct from a null reply. + var resp = "*0\r\n"; + + var result = Execute(resp, ResultProcessor.NullableRedisValueArray); + + Assert.NotNull(result); + Assert.Empty(result); + } + + [Fact] + public void NullArray_Resp2_IsNull() + { + // EXACTLY not satisfied: RESP2 null array. + var resp = "*-1\r\n"; + + var result = Execute(resp, ResultProcessor.NullableRedisValueArray); + + Assert.Null(result); + } + + [Fact] + public void Null_Resp3_IsNull() + { + // EXACTLY not satisfied: RESP3 null. + var resp = "_\r\n"; + + var result = Execute(resp, ResultProcessor.NullableRedisValueArray, protocol: RedisProtocol.Resp3); + + Assert.Null(result); + } + + [Fact] + public void Scalar_Failure() + { + // A bulk-string / scalar reply is not a valid LMOVEM response. + var resp = "$5\r\nhello\r\n"; + + ExecuteUnexpected(resp, ResultProcessor.NullableRedisValueArray); + } + + [Fact] + public void Integer_Failure() + { + var resp = ":5\r\n"; + + ExecuteUnexpected(resp, ResultProcessor.NullableRedisValueArray); + } +} diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/ListMoveMultipleRoundTrip.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/ListMoveMultipleRoundTrip.cs new file mode 100644 index 000000000..328fe04c6 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/ListMoveMultipleRoundTrip.cs @@ -0,0 +1,50 @@ +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests.RoundTripUnitTests; + +public class ListMoveMultipleRoundTrip(ITestOutputHelper log) +{ + // builds the message exactly as RedisDatabase.ListMove(... count ...) does, and asserts the wire bytes. + private static Message CreateMessage(ListSide from, ListSide to, long count, ListMoveCount mode, ListMoveOrder order) => + Message.Create( + 0, + CommandFlags.None, + RedisCommand.LMOVEM, + (RedisKey)"s", + (RedisKey)"d", + from.ToLiteral(), + to.ToLiteral(), + mode.ToLiteral(), + count, + order.ToLiteral()); + + [Fact(Timeout = 1000)] + public async Task UpTo_Bulk_RoundTrips() + { + var msg = CreateMessage(ListSide.Left, ListSide.Right, 2, ListMoveCount.UpTo, ListMoveOrder.Bulk); + const string requestResp = + "*8\r\n$6\r\nLMOVEM\r\n$1\r\ns\r\n$1\r\nd\r\n$4\r\nLEFT\r\n$5\r\nRIGHT\r\n$5\r\nCOUNT\r\n$1\r\n2\r\n$4\r\nBULK\r\n"; + + var result = await TestConnection.ExecuteAsync( + msg, ResultProcessor.NullableRedisValueArray, requestResp, "*2\r\n$1\r\na\r\n$1\r\nb\r\n", log: log); + + Assert.NotNull(result); + Assert.Equal(2, result.Length); + Assert.Equal("a", result[0].ToString()); + Assert.Equal("b", result[1].ToString()); + } + + [Fact(Timeout = 1000)] + public async Task Exactly_OneByOne_NotSatisfied_RoundTripsNull() + { + var msg = CreateMessage(ListSide.Right, ListSide.Left, 3, ListMoveCount.Exactly, ListMoveOrder.OneByOne); + const string requestResp = + "*8\r\n$6\r\nLMOVEM\r\n$1\r\ns\r\n$1\r\nd\r\n$5\r\nRIGHT\r\n$4\r\nLEFT\r\n$7\r\nEXACTLY\r\n$1\r\n3\r\n$3\r\nOBO\r\n"; + + var result = await TestConnection.ExecuteAsync( + msg, ResultProcessor.NullableRedisValueArray, requestResp, "*-1\r\n", log: log); + + Assert.Null(result); + } +} diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/SetCardinalityRoundTrip.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/SetCardinalityRoundTrip.cs new file mode 100644 index 000000000..500915437 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/SetCardinalityRoundTrip.cs @@ -0,0 +1,38 @@ +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests.RoundTripUnitTests; + +public class SetCardinalityRoundTrip(ITestOutputHelper log) +{ + [Fact(Timeout = 1000)] + public async Task SDiffCard_NoLimit_RoundTrips() + { + var msg = new SetOperationCardinalityMessage(0, CommandFlags.None, RedisCommand.SDIFFCARD, ["s1", "s2"], 0, approximate: false); + const string requestResp = "*4\r\n$9\r\nSDIFFCARD\r\n$1\r\n2\r\n$2\r\ns1\r\n$2\r\ns2\r\n"; + + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.Int64, requestResp, ":2\r\n", log: log); + Assert.Equal(2, result); + } + + [Fact(Timeout = 1000)] + public async Task SUnionCard_WithLimit_RoundTrips() + { + var msg = new SetOperationCardinalityMessage(0, CommandFlags.None, RedisCommand.SUNIONCARD, ["s1", "s2"], 3, approximate: false); + const string requestResp = "*6\r\n$10\r\nSUNIONCARD\r\n$1\r\n2\r\n$2\r\ns1\r\n$2\r\ns2\r\n$5\r\nLIMIT\r\n$1\r\n3\r\n"; + + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.Int64, requestResp, ":3\r\n", log: log); + Assert.Equal(3, result); + } + + [Fact(Timeout = 1000)] + public async Task SUnionCard_ApproxWithLimit_RoundTrips() + { + // APPROX is written before LIMIT + var msg = new SetOperationCardinalityMessage(0, CommandFlags.None, RedisCommand.SUNIONCARD, ["s1", "s2"], 3, approximate: true); + const string requestResp = "*7\r\n$10\r\nSUNIONCARD\r\n$1\r\n2\r\n$2\r\ns1\r\n$2\r\ns2\r\n$6\r\nAPPROX\r\n$5\r\nLIMIT\r\n$1\r\n3\r\n"; + + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.Int64, requestResp, ":3\r\n", log: log); + Assert.Equal(3, result); + } +} diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/StreamReadMaxCountRoundTrip.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/StreamReadMaxCountRoundTrip.cs new file mode 100644 index 000000000..9b4e204c8 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/StreamReadMaxCountRoundTrip.cs @@ -0,0 +1,34 @@ +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests.RoundTripUnitTests; + +public class StreamReadMaxCountRoundTrip(ITestOutputHelper log) +{ + [Fact(Timeout = 1000)] + public async Task XRead_CountMaxCountMaxSize_OrderedAfterCount() + { + StreamPosition[] positions = [new StreamPosition("sa", "5-5")]; + var msg = new RedisDatabase.MultiStreamReadCommandMessage(0, CommandFlags.None, positions, countPerStream: 2, maxCount: 3, maxSize: 100); + + // COUNT, then MAXCOUNT, then MAXSIZE, then STREAMS + const string requestResp = + "*10\r\n$5\r\nXREAD\r\n$5\r\nCOUNT\r\n$1\r\n2\r\n$8\r\nMAXCOUNT\r\n$1\r\n3\r\n$7\r\nMAXSIZE\r\n$3\r\n100\r\n$7\r\nSTREAMS\r\n$2\r\nsa\r\n$3\r\n5-5\r\n"; + + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.MultiStream, requestResp, "*-1\r\n", log: log); + Assert.Empty(result); + } + + [Fact(Timeout = 1000)] + public async Task XReadGroup_CountMaxCountMaxSize_OrderedAfterCount() + { + StreamPosition[] positions = [new StreamPosition("sa", "5-5")]; + var msg = new RedisDatabase.MultiStreamReadGroupCommandMessage(0, CommandFlags.None, positions, "g", "c", countPerStream: 2, noAck: false, claimMinIdleTime: null, maxCount: 3, maxSize: 100); + + const string requestResp = + "*13\r\n$10\r\nXREADGROUP\r\n$5\r\nGROUP\r\n$1\r\ng\r\n$1\r\nc\r\n$5\r\nCOUNT\r\n$1\r\n2\r\n$8\r\nMAXCOUNT\r\n$1\r\n3\r\n$7\r\nMAXSIZE\r\n$3\r\n100\r\n$7\r\nSTREAMS\r\n$2\r\nsa\r\n$3\r\n5-5\r\n"; + + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.MultiStream, requestResp, "*-1\r\n", log: log); + Assert.Empty(result); + } +} diff --git a/tests/StackExchange.Redis.Tests/SetTests.cs b/tests/StackExchange.Redis.Tests/SetTests.cs index 9326ca7a7..3040a54f3 100644 --- a/tests/StackExchange.Redis.Tests/SetTests.cs +++ b/tests/StackExchange.Redis.Tests/SetTests.cs @@ -124,6 +124,84 @@ public async Task SetIntersectionLengthAsync() Assert.Equal(0, await db.SetIntersectionLengthAsync([key3, key4])); } + [Fact] + public async Task SetCombineLength_Union() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + + var key1 = Me() + "1"; + db.KeyDelete(key1, CommandFlags.FireAndForget); + db.SetAdd(key1, [0, 1, 2, 3, 4], CommandFlags.FireAndForget); + var key2 = Me() + "2"; + db.KeyDelete(key2, CommandFlags.FireAndForget); + db.SetAdd(key2, [3, 4, 5], CommandFlags.FireAndForget); + + // union = {0,1,2,3,4,5} + Assert.Equal(6, db.SetCombineLength(SetOperation.Union, [key1, key2])); + Assert.Equal(6, await db.SetCombineLengthAsync(SetOperation.Union, [key1, key2])); + // with limit + Assert.Equal(4, db.SetCombineLength(SetOperation.Union, [key1, key2], 4)); + Assert.Equal(4, await db.SetCombineLengthAsync(SetOperation.Union, [key1, key2], 4)); + // approximate (HyperLogLog); exact for a set this small + Assert.Equal(6, db.SetCombineLength(SetOperation.Union, [key1, key2], approximate: true)); + Assert.Equal(4, await db.SetCombineLengthAsync(SetOperation.Union, [key1, key2], 4, approximate: true)); + + // Missing keys contribute nothing + var key3 = Me() + "3"; + db.KeyDelete(key3, CommandFlags.FireAndForget); + Assert.Equal(5, db.SetCombineLength(SetOperation.Union, [key1, key3])); + Assert.Equal(0, await db.SetCombineLengthAsync(SetOperation.Union, [key3])); + } + + [Fact] + public async Task SetCombineLength_Difference() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + + var key1 = Me() + "1"; + db.KeyDelete(key1, CommandFlags.FireAndForget); + db.SetAdd(key1, [0, 1, 2, 3, 4], CommandFlags.FireAndForget); + var key2 = Me() + "2"; + db.KeyDelete(key2, CommandFlags.FireAndForget); + db.SetAdd(key2, [3, 4, 5], CommandFlags.FireAndForget); + + // difference (key1 - key2) = {0,1,2} + Assert.Equal(3, db.SetCombineLength(SetOperation.Difference, [key1, key2])); + Assert.Equal(3, await db.SetCombineLengthAsync(SetOperation.Difference, [key1, key2])); + // with limit + Assert.Equal(2, db.SetCombineLength(SetOperation.Difference, [key1, key2], 2)); + Assert.Equal(2, await db.SetCombineLengthAsync(SetOperation.Difference, [key1, key2], 2)); + + // difference against a missing key leaves the first set intact + var key3 = Me() + "3"; + db.KeyDelete(key3, CommandFlags.FireAndForget); + Assert.Equal(5, db.SetCombineLength(SetOperation.Difference, [key1, key3])); + Assert.Equal(0, await db.SetCombineLengthAsync(SetOperation.Difference, [key3, key1])); + } + + [Fact] + public async Task SetCombineLength_Intersect() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + + var key1 = Me() + "1"; + db.KeyDelete(key1, CommandFlags.FireAndForget); + db.SetAdd(key1, [0, 1, 2, 3, 4], CommandFlags.FireAndForget); + var key2 = Me() + "2"; + db.KeyDelete(key2, CommandFlags.FireAndForget); + db.SetAdd(key2, [3, 4, 5], CommandFlags.FireAndForget); + + // intersect = {3,4}; SetCombineLength(Intersect) maps to SINTERCARD + Assert.Equal(2, db.SetCombineLength(SetOperation.Intersect, [key1, key2])); + Assert.Equal(1, await db.SetCombineLengthAsync(SetOperation.Intersect, [key1, key2], 1)); + } + [Fact] public async Task SScan() { diff --git a/tests/StackExchange.Redis.Tests/StreamTests.cs b/tests/StackExchange.Redis.Tests/StreamTests.cs index c3a7e6206..e2c9c4546 100644 --- a/tests/StackExchange.Redis.Tests/StreamTests.cs +++ b/tests/StackExchange.Redis.Tests/StreamTests.cs @@ -1280,6 +1280,70 @@ public async Task StreamConsumerGroupReadMultipleRestrictCount() Assert.Equal(id1_2, streams[0].Entries[0].Id); } + [Fact] + public async Task StreamReadMultipleMaxCount() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + var stream1 = Me() + "a"; + var stream2 = Me() + "b"; + + db.StreamAdd(stream1, "f", "v"); // 3 in stream1 + db.StreamAdd(stream1, "f", "v"); + db.StreamAdd(stream1, "f", "v"); + db.StreamAdd(stream2, "f", "v"); // 2 in stream2 + db.StreamAdd(stream2, "f", "v"); + + StreamPosition[] pairs = + [ + new StreamPosition(stream1, StreamPosition.Beginning), + new StreamPosition(stream2, StreamPosition.Beginning), + ]; + + // Without a global cap, all 5 come back. + var all = db.StreamRead(pairs, countPerStream: null); + Assert.Equal(5, all.Sum(s => s.Entries.Length)); + + // MAXCOUNT caps the *total* number of entries across all streams. + var capped = await db.StreamReadAsync(pairs, countPerStream: null, maxCount: 3); + Assert.Equal(3, capped.Sum(s => s.Entries.Length)); + + // MAXSIZE still returns at least one entry even with a tiny budget. + var oneish = db.StreamRead(pairs, countPerStream: null, maxSize: 1); + Assert.True(oneish.Sum(s => s.Entries.Length) >= 1); + } + + [Fact] + public async Task StreamReadGroupMultipleMaxCount() + { + await using var conn = Create(require: RedisFeatures.v8_10_0_rc2); + + var db = conn.GetDatabase(); + const string groupName = "test_group"; + var stream1 = Me() + "a"; + var stream2 = Me() + "b"; + + db.StreamAdd(stream1, "f", "v"); + db.StreamAdd(stream1, "f", "v"); + db.StreamAdd(stream1, "f", "v"); + db.StreamAdd(stream2, "f", "v"); + db.StreamAdd(stream2, "f", "v"); + + db.StreamCreateConsumerGroup(stream1, groupName, StreamPosition.Beginning); + db.StreamCreateConsumerGroup(stream2, groupName, StreamPosition.Beginning); + + StreamPosition[] pairs = + [ + new StreamPosition(stream1, StreamPosition.NewMessages), + new StreamPosition(stream2, StreamPosition.NewMessages), + ]; + + // MAXCOUNT caps the total across all streams (global budget of 3 vs the 5 available). + var capped = await db.StreamReadGroupAsync(pairs, groupName, "test_consumer", countPerStream: null, noAck: false, claimMinIdleTime: null, maxCount: 3); + Assert.Equal(3, capped.Sum(s => s.Entries.Length)); + } + [Fact] public async Task StreamConsumerGroupViewPendingInfoNoConsumers() {