Add support for blocking XREAD and XREADGROUP

gerzse · gerzse · commit f408902ea6fb · 2024-02-23T16:05:04.000+02:00
Issues redis#237 and redis#255.
diff --git a/src/NRedisStack/CoreCommands/CoreCommandBuilder.cs b/src/NRedisStack/CoreCommands/CoreCommandBuilder.cs
@@ -109,6 +109,81 @@ public static SerializedCommand BRPopLPush(RedisKey source, RedisKey destination
             return new SerializedCommand(RedisCoreCommands.BRPOPLPUSH, args);
         }
 
+        public static SerializedCommand XRead(RedisKey[] keys, RedisValue[] positions, int? count, int? timeoutMilliseconds)
+        {
+            if (keys.Length == 0)
+            {
+                throw new ArgumentException("At least one key must be provided.");
+            }
+
+            if (keys.Length != positions.Length)
+            {
+                throw new ArgumentException("The number of keys and positions must be the same.");
+            }
+
+            List<object> args = new List<object>();
+
+            if (count != null)
+            {
+                args.Add(CoreArgs.COUNT);
+                args.Add(count);
+            }
+
+            if (timeoutMilliseconds != null)
+            {
+                args.Add(CoreArgs.BLOCK);
+                args.Add(timeoutMilliseconds);
+            }
+
+            args.Add(CoreArgs.STREAMS);
+            args.AddRange(keys.Cast<object>());
+            args.AddRange(positions.Cast<object>());
+
+            return new SerializedCommand(RedisCoreCommands.XREAD, args);
+        }
+
+        public static SerializedCommand XReadGroup(RedisValue groupName, RedisValue consumerName, RedisKey[] keys, RedisValue[] positions, int? count, int? timeoutMilliseconds, bool? noAcknowledge)
+        {
+            if (keys.Length == 0)
+            {
+                throw new ArgumentException("At least one key must be provided.");
+            }
+
+            if (keys.Length != positions.Length)
+            {
+                throw new ArgumentException("The number of keys and positions must be the same.");
+            }
+
+            List<object> args = new List<object>();
+
+            args.Add(CoreArgs.GROUP);
+            args.Add(groupName);
+            args.Add(consumerName);
+
+            if (count != null)
+            {
+                args.Add(CoreArgs.COUNT);
+                args.Add(count);
+            }
+
+            if (timeoutMilliseconds != null)
+            {
+                args.Add(CoreArgs.BLOCK);
+                args.Add(timeoutMilliseconds);
+            }
+
+            if (noAcknowledge != null && noAcknowledge.Value)
+            {
+                args.Add(CoreArgs.NOACK);
+            }
+
+            args.Add(CoreArgs.STREAMS);
+            args.AddRange(keys.Cast<object>());
+            args.AddRange(positions.Cast<object>());
+
+            return new SerializedCommand(RedisCoreCommands.XREADGROUP, args);
+        }
+
         private static SerializedCommand BlockingCommandWithKeysAndTimeout(String command, RedisKey[] keys, double timeout)
         {
             if (keys.Length == 0)
diff --git a/src/NRedisStack/CoreCommands/CoreCommands.cs b/src/NRedisStack/CoreCommands/CoreCommands.cs
@@ -370,5 +370,116 @@ public static bool ClientSetInfo(this IDatabase db, SetInfoAttr attr, string val
             var command = CoreCommandBuilder.BRPopLPush(source, destination, timeout);
             return db.Execute(command).ToRedisValue();
         }
+
+        /// <summary>
+        /// The XREAD command.
+        /// <para/>
+        /// Read data from one or multiple streams, only returning entries with an ID greater than an ID provided by the caller.
+        /// </summary>
+        /// <param name="db">The <see cref="IDatabase"/> class where this extension method is applied.</param>
+        /// <param name="keys">Keys of the streams where to read from.</param>
+        /// <param name="positions">The positions from which to begin reading for each stream. See
+        /// <see cref="NRedisStack.Core.DataTypes.StreamSpecialIds"/> for special Ids that can be used.</param>
+        /// <param name="count">The maximum number of messages to return from each stream.</param>
+        /// <param name="timeoutMilliseconds">Amount of time in milliseconds to block in case all the streams are empty.
+        /// If not provided, or set to <c>null</c> then the read does not block. If set to <c>0</c> then it blocks indefinitely.</param>
+        /// <returns>A value of <see cref="RedisStreamEntries"/> for each stream, or <c>null</c> if the command times out
+        /// on the server.</returns>
+        /// <remarks>
+        /// <para>This is the blocking alternative for <seealso cref="IDatabase.StreamRead(StackExchange.Redis.StreamPosition[],System.Nullable{int},StackExchange.Redis.CommandFlags)"/>.</para>
+        /// <para><seealso href="https://redis.io/commands/xread"/></para>
+        /// </remarks>
+        public static RedisStreamEntries[]? XRead(this IDatabase db, RedisKey[] keys, RedisValue[] positions, int? count = null, int? timeoutMilliseconds = null)
+        {
+            var command = CoreCommandBuilder.XRead(keys, positions, count, timeoutMilliseconds);
+            return db.Execute(command).ToRedisStreamEntries();
+        }
+
+        /// <summary>
+        /// Syntactic sugar for <see cref="XRead(StackExchange.Redis.IDatabase,StackExchange.Redis.RedisKey[],StackExchange.Redis.RedisValue[],System.Nullable{int},System.Nullable{int})"/>,
+        /// where only one stream is being read from.
+        /// </summary>
+        /// <param name="db">The <see cref="IDatabase"/> class where this extension method is applied.</param>
+        /// <param name="key">Key of the stream where to read from.</param>
+        /// <param name="position">The position from which to begin reading. See
+        /// <see cref="NRedisStack.Core.DataTypes.StreamSpecialIds"/> for special Ids that can be used.</param>
+        /// <param name="count">The maximum number of messages to return from each stream.</param>
+        /// <param name="timeoutMilliseconds">Amount of time in milliseconds to block in case all the streams are empty.
+        /// If not provided, or set to <c>null</c> then the read does not block. If set to <c>0</c> then it blocks indefinitely.</param>
+        /// <returns>A <see cref="StreamEntry"/> list with the data read from the stream, of <c>null</c> if the command
+        /// times out on the server.</returns>
+        /// <remarks>
+        /// <para>This is the blocking alternative for <seealso cref="IDatabase.StreamRead(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,System.Nullable{int},StackExchange.Redis.CommandFlags)"/>.</para>
+        /// <para><seealso href="https://redis.io/commands/xread"/></para>
+        /// </remarks>
+        public static StreamEntry[]? XRead(this IDatabase db, RedisKey key, RedisValue position, int? count = null, int? timeoutMilliseconds = null)
+        {
+            var result = XRead(db, new[] { key }, new[] { position }, count, timeoutMilliseconds);
+            if (result == null || result.Length == 0)
+            {
+                return null;
+            }
+            return result[0].Entries;
+        }
+
+        /// <summary>
+        /// The XREADGROUP command.
+        /// <para/>
+        /// Read new or historical messages in one or several streams, for a consumer in a consumer group.
+        /// </summary>
+        /// <param name="db">The <see cref="IDatabase"/> class where this extension method is applied.</param>
+        /// <param name="groupName">The consumer group name.</param>
+        /// <param name="consumerName">The name of the consumer in the consumer group.</param>
+        /// <param name="keys">Keys of the streams where to read from.</param>
+        /// <param name="positions">The positions from which to begin reading for each stream. See
+        /// <see cref="NRedisStack.Core.DataTypes.StreamSpecialIds"/> for special Ids that can be used.</param>
+        /// <param name="count">The maximum number of messages to return from each stream.</param>
+        /// <param name="timeoutMilliseconds">Amount of time in milliseconds to block in case all the streams are empty.
+        /// If not provided, or set to <c>null</c> then the read does not block. If set to <c>0</c> then it blocks indefinitely.</param>
+        /// <param name="noAck">If set to <c>true</c> then inform the server that it should not wait for ACK for the
+        /// messages it sends to this read call.</param>
+        /// <returns>A value of <see cref="RedisStreamEntries"/> for each stream, or <c>null</c> if the command times out
+        /// on the server.</returns>
+        /// <remarks>
+        /// <para>This is the blocking alternative for <seealso cref="IDatabase.StreamReadGroup(StackExchange.Redis.StreamPosition[],StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Nullable{int},StackExchange.Redis.CommandFlags)"/>.</para>
+        /// <para><seealso href="https://redis.io/commands/xreadgroup"/></para>
+        /// </remarks>
+        public static RedisStreamEntries[]? XReadGroup(this IDatabase db, RedisValue groupName, RedisValue consumerName, RedisKey[] keys, RedisValue[] positions, int? count = null, int? timeoutMilliseconds = null, bool? noAck = null)
+        {
+            var command = CoreCommandBuilder.XReadGroup(groupName, consumerName, keys, positions, count, timeoutMilliseconds, noAck);
+            return db.Execute(command).ToRedisStreamEntries();
+        }
+
+        /// <summary>
+        /// Syntactic sugar for <see cref="XReadGroup(StackExchange.Redis.IDatabase,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisKey[],StackExchange.Redis.RedisValue[],System.Nullable{int},System.Nullable{int},System.Nullable{bool})"/>,
+        /// where only one stream is being read from.
+        /// </summary>
+        /// <param name="db">The <see cref="IDatabase"/> class where this extension method is applied.</param>
+        /// <param name="groupName">The consumer group name.</param>
+        /// <param name="consumerName">The name of the consumer in the consumer group.</param>
+        /// <param name="key">Key of the stream where to read from.</param>
+        /// <param name="position">The position from which to begin reading. See
+        /// <see cref="NRedisStack.Core.DataTypes.StreamSpecialIds"/> for special Ids that can be used.</param>
+        /// <param name="count">The maximum number of messages to return from each stream.</param>
+        /// <param name="timeoutMilliseconds">Amount of time in milliseconds to block in case all the streams are empty.
+        /// If not provided, or set to <c>null</c> then the read does not block. If set to <c>0</c> then it blocks indefinitely.</param>
+        /// <param name="noAck">If set to <c>true</c> then inform the server that it should not wait for ACK for the
+        /// messages it sends to this read call.</param>
+        /// <returns>A <see cref="StreamEntry"/> list with the data read from the stream, of <c>null</c> if the command
+        /// times out on the server.</returns>
+        /// <remarks>
+        /// <para>This is the blocking alternative for <seealso cref="IDatabase.StreamReadGroup(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Nullable{StackExchange.Redis.RedisValue},System.Nullable{int},StackExchange.Redis.CommandFlags)"/>.</para>
+        /// <para><seealso href="https://redis.io/commands/xreadgroup"/></para>
+        /// </remarks>
+        public static StreamEntry[]? XReadGroup(this IDatabase db, RedisValue groupName, RedisValue consumerName, RedisKey key, RedisValue position, int? count = null, int? timeoutMilliseconds = null, bool? noAck = null)
+        {
+            var result = XReadGroup(db, groupName, consumerName, new[] { key }, new[] { position }, count, timeoutMilliseconds, noAck);
+            if (result == null || result.Length == 0)
+            {
+                return null;
+            }
+            return result[0].Entries;
+        }
+
     }
 }
diff --git a/src/NRedisStack/CoreCommands/DataTypes/RedisStreamEntries.cs b/src/NRedisStack/CoreCommands/DataTypes/RedisStreamEntries.cs
@@ -0,0 +1,25 @@
+using StackExchange.Redis;
+
+namespace NRedisStack.Core.DataTypes;
+
+/// <summary>
+/// Holds the key and the entries for a Redis Stream, as returned by, for example, the XREAD or the XREADGROUP commands.
+/// </summary>
+public readonly struct RedisStreamEntries
+{
+    internal RedisStreamEntries(RedisKey key, StreamEntry[] entries)
+    {
+        Key = key;
+        Entries = entries;
+    }
+
+    /// <summary>
+    /// The key for the stream.
+    /// </summary>
+    public RedisKey Key { get; }
+
+    /// <summary>
+    /// An array of entries contained within the stream.
+    /// </summary>
+    public StreamEntry[] Entries { get; }
+}
diff --git a/src/NRedisStack/CoreCommands/DataTypes/StreamSpecialIds.cs b/src/NRedisStack/CoreCommands/DataTypes/StreamSpecialIds.cs
@@ -0,0 +1,22 @@
+namespace NRedisStack.Core.DataTypes;
+
+/// <summary>
+/// Constants for special stream Ids, to be used, for example, with the XREAD and XREADGROUP commands
+/// </summary>
+public class StreamSpecialIds
+{
+    /// <summary>
+    /// Smallest incomplete ID, can be used for reading from the very first message in a stream.
+    /// </summary>
+    public const string AllMessagesId = "0";
+
+    /// <summary>
+    /// For receiving only new messages that arrive after blocking on a read.
+    /// </summary>
+    public const string NewMessagesId = "$";
+
+    /// <summary>
+    /// For receiving only messages that were never delivered to any other consumer.
+    /// </summary>
+    public const string UndeliveredMessagesId = ">";
+}
diff --git a/src/NRedisStack/CoreCommands/Literals/CommandArgs.cs b/src/NRedisStack/CoreCommands/Literals/CommandArgs.cs
@@ -2,11 +2,15 @@ namespace NRedisStack.Core.Literals
 {
     internal static class CoreArgs
     {
+        public const string BLOCK = "BLOCK";
         public const string COUNT = "COUNT";
+        public const string GROUP = "GROUP";
         public const string LEFT = "LEFT";
         public const string MAX = "MAX";
         public const string MIN = "MIN";
+        public const string NOACK = "NOACK";
         public const string RIGHT = "RIGHT";
+        public const string STREAMS = "STREAMS";
         public const string lib_name = "LIB-NAME";
         public const string lib_ver = "LIB-VER";
     }
diff --git a/src/NRedisStack/CoreCommands/Literals/Commands.cs b/src/NRedisStack/CoreCommands/Literals/Commands.cs
@@ -15,5 +15,7 @@ internal static class RedisCoreCommands
         public const string BZPOPMIN = "BZPOPMIN";
         public const string CLIENT = "CLIENT";
         public const string SETINFO = "SETINFO";
+        public const string XREAD = "XREAD";
+        public const string XREADGROUP = "XREADGROUP";
     }
 }
diff --git a/src/NRedisStack/ResponseParser.cs b/src/NRedisStack/ResponseParser.cs
@@ -802,5 +802,56 @@ public static Dictionary<string, RedisResult>[] ToDictionarys(this RedisResult r
 
             return new Tuple<RedisKey, List<RedisValue>>(resultKey, values);
         }
+
+        public static RedisStreamEntries[]? ToRedisStreamEntries(this RedisResult result)
+        {
+            if (result.IsNull)
+            {
+                return null;
+            }
+
+            var resultArray = (RedisResult[])result!;
+            RedisStreamEntries[] redisStreamEntries = new RedisStreamEntries[resultArray.Length];
+            for (int i = 0; i < resultArray.Length; i++)
+            {
+                RedisResult[] streamResultArray = (RedisResult[])resultArray[i]!;
+                RedisKey streamKey = streamResultArray[0].ToRedisKey();
+                StreamEntry[] streamEntries = ParseStreamEntries(streamResultArray[1].ToArray());
+                redisStreamEntries[i] = new RedisStreamEntries(streamKey, streamEntries);
+            }
+
+            return redisStreamEntries;
+        }
+
+        private static StreamEntry[] ParseStreamEntries(IReadOnlyList<RedisResult> results)
+        {
+            int count = results.Count;
+            StreamEntry[] streamEntries = new StreamEntry[count];
+
+            for (int i = 0; i < count; i++)
+            {
+                RedisResult[] streamEntryArray = (RedisResult[])results[i]!;
+                RedisValue key = streamEntryArray[0].ToRedisValue();
+                NameValueEntry[] nameValueEntries = ParseNameValueEntries(streamEntryArray[1].ToArray());
+                streamEntries[i] = new StreamEntry(key, nameValueEntries);
+            }
+
+            return streamEntries;
+        }
+
+        private static NameValueEntry[] ParseNameValueEntries(IReadOnlyList<RedisResult> redisResults)
+        {
+            int count = redisResults.Count / 2;
+            var nameValueEntries = new NameValueEntry[count];
+
+            for (int i = 0; i < count; i++)
+            {
+                nameValueEntries[i] = new NameValueEntry(
+                    redisResults[2 * i].ToRedisValue(),
+                    redisResults[2 * i + 1].ToRedisValue());
+            }
+
+            return nameValueEntries;
+        }
     }
 }
diff --git a/tests/NRedisStack.Tests/Core Commands/CoreTests.cs b/tests/NRedisStack.Tests/Core Commands/CoreTests.cs

Original file line number	Diff line number	Diff line change
`@@ -2,11 +2,15 @@ namespace NRedisStack.Core.Literals`
`2`	`2`	`{`
`3`	`3`	`internal static class CoreArgs`
`4`	`4`	`{`
	`5`	`+ public const string BLOCK = "BLOCK";`
`5`	`6`	`public const string COUNT = "COUNT";`
	`7`	`+ public const string GROUP = "GROUP";`
`6`	`8`	`public const string LEFT = "LEFT";`
`7`	`9`	`public const string MAX = "MAX";`
`8`	`10`	`public const string MIN = "MIN";`
	`11`	`+ public const string NOACK = "NOACK";`
`9`	`12`	`public const string RIGHT = "RIGHT";`
	`13`	`+ public const string STREAMS = "STREAMS";`
`10`	`14`	`public const string lib_name = "LIB-NAME";`
`11`	`15`	`public const string lib_ver = "LIB-VER";`
`12`	`16`	`}`
Original file line number	Diff line number	Diff line change
`@@ -15,5 +15,7 @@ internal static class RedisCoreCommands`
`15`	`15`	`public const string BZPOPMIN = "BZPOPMIN";`
`16`	`16`	`public const string CLIENT = "CLIENT";`
`17`	`17`	`public const string SETINFO = "SETINFO";`
	`18`	`+ public const string XREAD = "XREAD";`
	`19`	`+ public const string XREADGROUP = "XREADGROUP";`
`18`	`20`	`}`
`19`	`21`	`}`