Support the XAUTOCLAIM command.

docs/ReleaseNotes.md

@@ -23,6 +23,7 @@
 - Adds: Support for `GEOSEARCH` with `.GeoSearch()`/`.GeoSearchAsync()` ([#2089 by slorello89](https://github.com/StackExchange/StackExchange.Redis/pull/2089))
 - Adds: Support for `GEOSEARCHSTORE` with `.GeoSearchAndStore()`/`.GeoSearchAndStoreAsync()` ([#2089 by slorello89](https://github.com/StackExchange/StackExchange.Redis/pull/2089))
 - Adds: Support for `HRANDFIELD` with `.HashRandomField()`/`.HashRandomFieldAsync()`, `.HashRandomFields()`/`.HashRandomFieldsAsync()`, and `.HashRandomFieldsWithValues()`/`.HashRandomFieldsWithValuesAsync()` ([#2090 by slorello89](https://github.com/StackExchange/StackExchange.Redis/pull/2090))
+- Adds: Support for `XAUTOCLAIM` with `.StreamAutoClaim()`/.`StreamAutoClaimAsync()` and `.StreamAutoClaimIdsOnly()`/.`StreamAutoClaimIdsOnlyAsync()` ([#2095 by ttingen](https://github.com/StackExchange/StackExchange.Redis/pull/2095))
 
 ## 2.5.61
 

src/StackExchange.Redis/APITypes/StreamAutoClaimIdsOnlyResult.cs

@@ -0,0 +1,41 @@
+using System;
+
+namespace StackExchange.Redis;
+
+/// <summary>
+/// Result of the <see href="https://redis.io/commands/xautoclaim/">XAUTOCLAIM</see> command with the <c>JUSTID</c> option.
+/// </summary>
+public readonly struct StreamAutoClaimIdsOnlyResult
+{
+    internal StreamAutoClaimIdsOnlyResult(RedisValue nextStartId, RedisValue[] claimedIds, RedisValue[] deletedIds)
+    {
+        NextStartId = nextStartId;
+        ClaimedIds = claimedIds;
+        DeletedIds = deletedIds;
+    }
+
+    /// <summary>
+    /// A null <see cref="StreamAutoClaimIdsOnlyResult"/>, indicating no results.
+    /// </summary>
+    public static StreamAutoClaimIdsOnlyResult Null { get; } = new StreamAutoClaimIdsOnlyResult(RedisValue.Null, Array.Empty<RedisValue>(), Array.Empty<RedisValue>());
+
+    /// <summary>
+    /// Whether this object is null/empty.
+    /// </summary>
+    public bool IsNull => NextStartId.IsNull && ClaimedIds == Array.Empty<RedisValue>() && DeletedIds == Array.Empty<RedisValue>();
+
+    /// <summary>
+    /// The stream ID to be used in the next call to StreamAutoClaim.
+    /// </summary>
+    public RedisValue NextStartId { get; }
+
+    /// <summary>
+    /// Array of IDs claimed by the command.
+    /// </summary>
+    public RedisValue[] ClaimedIds { get; }
+
+    /// <summary>
+    /// Array of message IDs deleted from the stream.
+    /// </summary>
+    public RedisValue[] DeletedIds { get; }
+}

src/StackExchange.Redis/APITypes/StreamAutoClaimResult.cs

@@ -0,0 +1,41 @@
+using System;
+
+namespace StackExchange.Redis;
+
+/// <summary>
+/// Result of the <see href="https://redis.io/commands/xautoclaim/">XAUTOCLAIM</see> command.
+/// </summary>
+public readonly struct StreamAutoClaimResult
+{
+    internal StreamAutoClaimResult(RedisValue nextStartId, StreamEntry[] claimedEntries, RedisValue[] deletedIds)
+    {
+        NextStartId = nextStartId;
+        ClaimedEntries = claimedEntries;
+        DeletedIds = deletedIds;
+    }
+
+    /// <summary>
+    /// A null <see cref="StreamAutoClaimResult"/>, indicating no results.
+    /// </summary>
+    public static StreamAutoClaimResult Null { get; } = new StreamAutoClaimResult(RedisValue.Null, Array.Empty<StreamEntry>(), Array.Empty<RedisValue>());
+
+    /// <summary>
+    /// Whether this object is null/empty.
+    /// </summary>
+    public bool IsNull => NextStartId.IsNull && ClaimedEntries == Array.Empty<StreamEntry>() && DeletedIds == Array.Empty<RedisValue>();
+
+    /// <summary>
+    /// The stream ID to be used in the next call to StreamAutoClaim.
+    /// </summary>
+    public RedisValue NextStartId { get; }
+
+    /// <summary>
+    /// An array of <see cref="StreamEntry"/> for the successfully claimed entries.
+    /// </summary>
+    public StreamEntry[] ClaimedEntries { get; }
+
+    /// <summary>
+    /// An array of message IDs deleted from the stream.
+    /// </summary>
+    public RedisValue[] DeletedIds { get; }
+}

src/StackExchange.Redis/Enums/RedisCommand.cs

@@ -189,6 +189,7 @@ internal enum RedisCommand
 
     XACK,
     XADD,
+    XAUTOCLAIM,
     XCLAIM,
     XDEL,
     XGROUP,
@@ -324,6 +325,7 @@ internal static bool IsPrimaryOnly(this RedisCommand command)
             case RedisCommand.UNLINK:
             case RedisCommand.XACK:
             case RedisCommand.XADD:
+            case RedisCommand.XAUTOCLAIM:
             case RedisCommand.XCLAIM:
             case RedisCommand.XDEL:
             case RedisCommand.XGROUP:

src/StackExchange.Redis/Interfaces/IDatabase.cs

@@ -2163,6 +2163,37 @@ IEnumerable<SortedSetEntry> SortedSetScan(RedisKey key,
         /// <remarks><seealso href="https://redis.io/commands/xadd"/></remarks>
         RedisValue StreamAdd(RedisKey key, NameValueEntry[] streamPairs, RedisValue? messageId = null, int? maxLength = null, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None);
 
+        /// <summary>
+        /// Change ownership of messages consumed, but not yet acknowledged, by a different consumer.
+        /// Messages that have been idle for more than <paramref name="minIdleTimeInMs"/> will be claimed.
+        /// </summary>
+        /// <param name="key">The key of the stream.</param>
+        /// <param name="consumerGroup">The consumer group.</param>
+        /// <param name="claimingConsumer">The consumer claiming the messages(s).</param>
+        /// <param name="minIdleTimeInMs">The minimum idle time threshold for pending messages to be claimed.</param>
+        /// <param name="startAtId">The starting ID to scan for pending messages that have an idle time greater than <paramref name="minIdleTimeInMs"/>.</param>
+        /// <param name="count">The upper limit of the number of entries that the command attempts to claim. If <see langword="null"/>, Redis will default the value to 100.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>An instance of <see cref="StreamAutoClaimResult"/>.</returns>
+        /// <remarks><seealso href="https://redis.io/commands/xautoclaim"/></remarks>
+        StreamAutoClaimResult StreamAutoClaim(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None);
+
+        /// <summary>
+        /// Change ownership of messages consumed, but not yet acknowledged, by a different consumer.
+        /// Messages that have been idle for more than <paramref name="minIdleTimeInMs"/> will be claimed.
+        /// The result will contain the claimed message IDs instead of a <see cref="StreamEntry"/> instance.
+        /// </summary>
+        /// <param name="key">The key of the stream.</param>
+        /// <param name="consumerGroup">The consumer group.</param>
+        /// <param name="claimingConsumer">The consumer claiming the messages(s).</param>
+        /// <param name="minIdleTimeInMs">The minimum idle time threshold for pending messages to be claimed.</param>
+        /// <param name="startAtId">The starting ID to scan for pending messages that have an idle time greater than <paramref name="minIdleTimeInMs"/>.</param>
+        /// <param name="count">The upper limit of the number of entries that the command attempts to claim. If <see langword="null"/>, Redis will default the value to 100.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>An instance of <see cref="StreamAutoClaimIdsOnlyResult"/>.</returns>
+        /// <remarks><seealso href="https://redis.io/commands/xautoclaim"/></remarks>
+        StreamAutoClaimIdsOnlyResult StreamAutoClaimIdsOnly(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Change ownership of messages consumed, but not yet acknowledged, by a different consumer.
         /// This method returns the complete message for the claimed message(s).

src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs

@@ -2116,6 +2116,37 @@ IAsyncEnumerable<SortedSetEntry> SortedSetScanAsync(RedisKey key,
         /// <remarks><seealso href="https://redis.io/commands/xadd"/></remarks>
         Task<RedisValue> StreamAddAsync(RedisKey key, NameValueEntry[] streamPairs, RedisValue? messageId = null, int? maxLength = null, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None);
 
+        /// <summary>
+        /// Change ownership of messages consumed, but not yet acknowledged, by a different consumer.
+        /// Messages that have been idle for more than <paramref name="minIdleTimeInMs"/> will be claimed.
+        /// </summary>
+        /// <param name="key">The key of the stream.</param>
+        /// <param name="consumerGroup">The consumer group.</param>
+        /// <param name="claimingConsumer">The consumer claiming the messages(s).</param>
+        /// <param name="minIdleTimeInMs">The minimum idle time threshold for pending messages to be claimed.</param>
+        /// <param name="startAtId">The starting ID to scan for pending messages that have an idle time greater than <paramref name="minIdleTimeInMs"/>.</param>
+        /// <param name="count">The upper limit of the number of entries that the command attempts to claim. If <see langword="null"/>, Redis will default the value to 100.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>An instance of <see cref="StreamAutoClaimResult"/>.</returns>
+        /// <remarks><seealso href="https://redis.io/commands/xautoclaim"/></remarks>
+        Task<StreamAutoClaimResult> StreamAutoClaimAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None);
+
+        /// <summary>
+        /// Change ownership of messages consumed, but not yet acknowledged, by a different consumer.
+        /// Messages that have been idle for more than <paramref name="minIdleTimeInMs"/> will be claimed.
+        /// The result will contain the claimed message IDs instead of a <see cref="StreamEntry"/> instance.
+        /// </summary>
+        /// <param name="key">The key of the stream.</param>
+        /// <param name="consumerGroup">The consumer group.</param>
+        /// <param name="claimingConsumer">The consumer claiming the messages(s).</param>
+        /// <param name="minIdleTimeInMs">The minimum idle time threshold for pending messages to be claimed.</param>
+        /// <param name="startAtId">The starting ID to scan for pending messages that have an idle time greater than <paramref name="minIdleTimeInMs"/>.</param>
+        /// <param name="count">The upper limit of the number of entries that the command attempts to claim. If <see langword="null"/>, Redis will default the value to 100.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>An instance of <see cref="StreamAutoClaimIdsOnlyResult"/>.</returns>
+        /// <remarks><seealso href="https://redis.io/commands/xautoclaim"/></remarks>
+        Task<StreamAutoClaimIdsOnlyResult> StreamAutoClaimIdsOnlyAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Change ownership of messages consumed, but not yet acknowledged, by a different consumer.
         /// This method returns the complete message for the claimed message(s).

src/StackExchange.Redis/KeyspaceIsolation/DatabaseWrapper.cs

@@ -496,6 +496,12 @@ public RedisValue StreamAdd(RedisKey key, RedisValue streamField, RedisValue str
         public RedisValue StreamAdd(RedisKey key, NameValueEntry[] streamPairs, RedisValue? messageId = null, int? maxLength = null, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None) =>
             Inner.StreamAdd(ToInner(key), streamPairs, messageId, maxLength, useApproximateMaxLength, flags);
 
+        public StreamAutoClaimResult StreamAutoClaim(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None) =>
+            Inner.StreamAutoClaim(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, startAtId, count, flags);
+
+        public StreamAutoClaimIdsOnlyResult StreamAutoClaimIdsOnly(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None) =>
+            Inner.StreamAutoClaimIdsOnly(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, startAtId, count, flags);
+
         public StreamEntry[] StreamClaim(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None) =>
             Inner.StreamClaim(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
 

src/StackExchange.Redis/KeyspaceIsolation/WrapperBase.cs

@@ -513,6 +513,12 @@ public Task<RedisValue> StreamAddAsync(RedisKey key, RedisValue streamField, Red
         public Task<RedisValue> StreamAddAsync(RedisKey key, NameValueEntry[] streamPairs, RedisValue? messageId = null, int? maxLength = null, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None) =>
             Inner.StreamAddAsync(ToInner(key), streamPairs, messageId, maxLength, useApproximateMaxLength, flags);
 
+        public Task<StreamAutoClaimResult> StreamAutoClaimAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None) =>
+            Inner.StreamAutoClaimAsync(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, startAtId, count, flags);
+
+        public Task<StreamAutoClaimIdsOnlyResult> StreamAutoClaimIdsOnlyAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue startAtId, int? count = null, CommandFlags flags = CommandFlags.None) =>
+            Inner.StreamAutoClaimIdsOnlyAsync(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, startAtId, count, flags);
+
         public Task<StreamEntry[]> StreamClaimAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None) =>
             Inner.StreamClaimAsync(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);