Eliminate the need for Stream magic strings & added additional XGROUP… (#878)

* Eliminate the need for Stream magic strings & added additional XGROUP subcommands.

- Replaced the need for magic strings with Range, ReadOffset, and GroupReadOffset structs.
- Condensed some of the method signatures due to the new structs.
- Added methods & tests for two of the XGROUP subcommands (DESTROY & DELCONSUMER).

* Updated tests with their original IDs used.

* Use a single struct to represent stream position for Group, Read, and ReadGroup commands.

- Consolidated GroupCreateOptions, GroupReadOffset, and ReadOffset into a single struct, Position.
- Removed the "Pair" structs and created StreamPosition, this will be used in the multi-stream read commands.
- Renamed RedisStreamEntry to StreamEntry.
- Also added the remaining subcommand for the XGROUP command (SETID) and an associated unit test.

* File cleanup, delete consolidated structs.

* Additional changes and unit tests for the Position struct.

* Update PositionKind enum with explicit zero value.

StackExchange.Redis.Tests/DatabaseWrapperTests.cs

@@ -828,6 +828,13 @@ public void StreamClaimMessagesReturningIds()
             mock.Verify(_ => _.StreamClaimIdsOnly("prefix:key", "group", "consumer", 1000, messageIds, CommandFlags.HighPriority));
         }
 
+        [Fact]
+        public void StreamConsumerGroupSetPosition()
+        {
+            wrapper.StreamConsumerGroupSetPosition("key", "group", Position.Beginning, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamConsumerGroupSetPosition("prefix:key", "group", Position.Beginning, CommandFlags.HighPriority));
+        }
+
         [Fact]
         public void StreamConsumerInfoGet()
         {
@@ -838,8 +845,8 @@ public void StreamConsumerInfoGet()
         [Fact]
         public void StreamCreateConsumerGroup()
         {
-            wrapper.StreamCreateConsumerGroup("key", "group", "0-0", CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamCreateConsumerGroup("prefix:key", "group", "0-0", CommandFlags.HighPriority));
+            wrapper.StreamCreateConsumerGroup("key", "group", Position.Beginning, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamCreateConsumerGroup("prefix:key", "group", Position.Beginning, CommandFlags.HighPriority));
         }
 
         [Fact]
@@ -871,6 +878,20 @@ public void StreamMessagesDelete()
             mock.Verify(_ => _.StreamDelete("prefix:key", messageIds, CommandFlags.HighPriority));
         }
 
+        [Fact]
+        public void StreamDeleteConsumer()
+        {
+            wrapper.StreamDeleteConsumer("key", "group", "consumer", CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamDeleteConsumer("prefix:key", "group", "consumer", CommandFlags.HighPriority));
+        }
+
+        [Fact]
+        public void StreamDeleteConsumerGroup()
+        {
+            wrapper.StreamDeleteConsumerGroup("key", "group", CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamDeleteConsumerGroup("prefix:key", "group", CommandFlags.HighPriority));
+        }
+
         [Fact]
         public void StreamPendingInfoGet()
         {
@@ -881,37 +902,45 @@ public void StreamPendingInfoGet()
         [Fact]
         public void StreamPendingMessageInfoGet()
         {
-            wrapper.StreamPendingMessages("key", "group", 10, RedisValue.Null, null, null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamPendingMessages("prefix:key", "group", 10, RedisValue.Null, null, null, CommandFlags.HighPriority));
+            wrapper.StreamPendingMessages("key", "group", 10, RedisValue.Null, "-", "+", CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamPendingMessages("prefix:key", "group", 10, RedisValue.Null, "-", "+", CommandFlags.HighPriority));
         }
 
         [Fact]
         public void StreamRange()
         {
             wrapper.StreamRange("key", "-", "+", null, Order.Ascending, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamRange("prefix:key", "-", "+",null, Order.Ascending, CommandFlags.HighPriority));
+            mock.Verify(_ => _.StreamRange("prefix:key", "-", "+", null, Order.Ascending, CommandFlags.HighPriority));
         }
 
         [Fact]
         public void StreamRead_1()
         {
-            var keysAndIds = new StreamIdPair[0] { };
-            wrapper.StreamRead(keysAndIds, null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamRead(keysAndIds, null, CommandFlags.HighPriority));
+            var streamPositions = new StreamPosition[0] { };
+            wrapper.StreamRead(streamPositions, null, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamRead(streamPositions, null, CommandFlags.HighPriority));
         }
 
         [Fact]
         public void StreamRead_2()
         {
-            wrapper.StreamRead("key", "0-0", null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamRead("prefix:key", "0-0", null, CommandFlags.HighPriority));
+            wrapper.StreamRead("key", new Position("0-0"), null, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamRead("prefix:key", new Position("0-0"), null, CommandFlags.HighPriority));
+        }
+
+        [Fact]
+        public void StreamStreamReadGroup_1()
+        {
+            wrapper.StreamReadGroup("key", "group", "consumer", new Position("0-0"), 10, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadGroup("prefix:key", "group", "consumer", new Position("0-0"), 10, CommandFlags.HighPriority));
         }
 
         [Fact]
-        public void StreamStreamReadGroup()
+        public void StreamStreamReadGroup_2()
         {
-            wrapper.StreamReadGroup("key", "group", "consumer", "0-0", 10, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamReadGroup("prefix:key", "group", "consumer", "0-0", 10, CommandFlags.HighPriority));
+            var streamPositions = new StreamPosition[0] { };
+            wrapper.StreamReadGroup(streamPositions, "group", "consumer", 10, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadGroup(streamPositions, "group", "consumer", 10, CommandFlags.HighPriority));
         }
 
         [Fact]

StackExchange.Redis.Tests/WrapperBaseTests.cs

@@ -793,11 +793,18 @@ public void StreamConsumerInfoGetAsync()
             mock.Verify(_ => _.StreamConsumerInfoAsync("prefix:key", "group", CommandFlags.HighPriority));
         }
 
+        [Fact]
+        public void StreamConsumerGroupSetPositionAsync()
+        {
+            wrapper.StreamConsumerGroupSetPositionAsync("key", "group", Position.Beginning, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamConsumerGroupSetPositionAsync("prefix:key", "group", Position.Beginning, CommandFlags.HighPriority));
+        }
+
         [Fact]
         public void StreamCreateConsumerGroupAsync()
         {
-            wrapper.StreamCreateConsumerGroupAsync("key", "group", "0-0", CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamCreateConsumerGroupAsync("prefix:key", "group", "0-0", CommandFlags.HighPriority));
+            wrapper.StreamCreateConsumerGroupAsync("key", "group", new Position("0-0"), CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamCreateConsumerGroupAsync("prefix:key", "group", new Position("0-0"), CommandFlags.HighPriority));
         }
 
         [Fact]
@@ -829,6 +836,20 @@ public void StreamMessagesDeleteAsync()
             mock.Verify(_ => _.StreamDeleteAsync("prefix:key", messageIds, CommandFlags.HighPriority));
         }
 
+        [Fact]
+        public void StreamDeleteConsumerAsync()
+        {
+            wrapper.StreamDeleteConsumerAsync("key", "group", "consumer", CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamDeleteConsumerAsync("prefix:key", "group", "consumer", CommandFlags.HighPriority));
+        }
+
+        [Fact]
+        public void StreamDeleteConsumerGroupAsync()
+        {
+            wrapper.StreamDeleteConsumerGroupAsync("key", "group", CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamDeleteConsumerGroupAsync("prefix:key", "group", CommandFlags.HighPriority));
+        }
+
         [Fact]
         public void StreamPendingInfoGetAsync()
         {
@@ -853,23 +874,31 @@ public void StreamRangeAsync()
         [Fact]
         public void StreamReadAsync_1()
         {
-            var keysAndIds = new StreamIdPair[0] { };
-            wrapper.StreamReadAsync(keysAndIds, null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamReadAsync(keysAndIds, null, CommandFlags.HighPriority));
+            var streamPositions = new StreamPosition[0] { };
+            wrapper.StreamReadAsync(streamPositions, null, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadAsync(streamPositions, null, CommandFlags.HighPriority));
         }
 
         [Fact]
         public void StreamReadAsync_2()
         {
-            wrapper.StreamReadAsync("key", "0-0", null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamReadAsync("prefix:key", "0-0", null, CommandFlags.HighPriority));
+            wrapper.StreamReadAsync("key", new Position("0-0"), null, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadAsync("prefix:key", new Position("0-0"), null, CommandFlags.HighPriority));
+        }
+
+        [Fact]
+        public void StreamReadGroupAsync_1()
+        {
+            wrapper.StreamReadGroupAsync("key", "group", "consumer", Position.Beginning, 10, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadGroupAsync("prefix:key", "group", "consumer", Position.Beginning, 10, CommandFlags.HighPriority));
         }
 
         [Fact]
-        public void StreamReadGroupAsync()
+        public void StreamStreamReadGroupAsync_2()
         {
-            wrapper.StreamReadGroupAsync("key", "group", "consumer", "0-0", 10, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamReadGroupAsync("prefix:key", "group", "consumer", "0-0", 10, CommandFlags.HighPriority));
+            var streamPositions = new StreamPosition[0] { };
+            wrapper.StreamReadGroupAsync(streamPositions, "group", "consumer", 10, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadGroupAsync(streamPositions, "group", "consumer", 10, CommandFlags.HighPriority));
         }
 
         [Fact]

StackExchange.Redis/StackExchange/Redis/Enums/PositionKind.cs

+namespace StackExchange.Redis
+{
+    internal enum PositionKind
+    {
+        Beginning = 0,
+        Explicit = 1,
+        New = 2
+    }
+}

StackExchange.Redis/StackExchange/Redis/KeyspaceIsolation/DatabaseWrapper.cs

@@ -607,7 +607,7 @@ public RedisValue StreamAdd(RedisKey key, NameValueEntry[] streamPairs, RedisVal
             return Inner.StreamAdd(ToInner(key), streamPairs, messageId, maxLength, useApproximateMaxLength, flags);
         }
 
-        public RedisStreamEntry[] StreamClaim(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamClaim(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamClaim(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
         }
@@ -617,9 +617,14 @@ public RedisValue[] StreamClaimIdsOnly(RedisKey key, RedisValue consumerGroup, R
             return Inner.StreamClaimIdsOnly(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
         }
 
-        public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, RedisValue? readFrom = null, CommandFlags flags = CommandFlags.None)
+        public bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamCreateConsumerGroup(ToInner(key), groupName, readFrom, flags);
+            return Inner.StreamConsumerGroupSetPosition(ToInner(key), groupName, position, flags);
+        }
+
+        public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamCreateConsumerGroup(ToInner(key), groupName, position, flags);
         }
 
         public StreamInfo StreamInfo(RedisKey key, CommandFlags flags = CommandFlags.None)
@@ -647,6 +652,16 @@ public long StreamDelete(RedisKey key, RedisValue[] messageIds, CommandFlags fla
             return Inner.StreamDelete(ToInner(key), messageIds, flags);
         }
 
+        public long StreamDeleteConsumer(RedisKey key, RedisValue groupName, RedisValue consumerName, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamDeleteConsumer(ToInner(key), groupName, consumerName, flags);
+        }
+
+        public bool StreamDeleteConsumerGroup(RedisKey key, RedisValue groupName, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamDeleteConsumerGroup(ToInner(key), groupName, flags);
+        }
+
         public StreamPendingInfo StreamPending(RedisKey key, RedisValue groupName, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamPending(ToInner(key), groupName, flags);
@@ -657,24 +672,29 @@ public StreamPendingMessageInfo[] StreamPendingMessages(RedisKey key, RedisValue
             return Inner.StreamPendingMessages(ToInner(key), groupName, count, consumerName, minId, maxId, flags);
         }
 
-        public RedisStreamEntry[] StreamRange(RedisKey key, RedisValue? minId = null, RedisValue? maxId = null, int? count = null, Order order = Order.Ascending, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamRange(RedisKey key, RedisValue? minId = null, RedisValue? maxId = null, int? count = null, Order messageOrder = Order.Ascending, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamRange(ToInner(key), minId, maxId, count, messageOrder, flags);
+        }
+
+        public StreamEntry[] StreamRead(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamRange(ToInner(key), minId, maxId, count, order, flags);
+            return Inner.StreamRead(ToInner(key), position, count, flags);
         }
 
-        public RedisStreamEntry[] StreamRead(RedisKey key, RedisValue afterId, int? count = null, CommandFlags flags = CommandFlags.None)
+        public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamRead(ToInner(key), afterId, count, flags);
+            return Inner.StreamRead(streamPositions, countPerStream, flags);
         }
 
-        public RedisStream[] StreamRead(StreamIdPair[] streamIdPairs, int? countPerStream = null, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamRead(streamIdPairs, countPerStream, flags);
+            return Inner.StreamReadGroup(ToInner(key), groupName, consumerName, position, count, flags);
         }
 
-        public RedisStreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? readFromId = null, int? count = null, CommandFlags flags = CommandFlags.None)
+        public RedisStream[] StreamReadGroup(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamReadGroup(ToInner(key), groupName, consumerName, readFromId, count, flags);
+            return Inner.StreamReadGroup(streamPositions, groupName, consumerName, countPerStream, flags);
         }
 
         public long StreamTrim(RedisKey key, int maxLength, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None)

StackExchange.Redis/StackExchange/Redis/KeyspaceIsolation/WrapperBase.cs

@@ -586,7 +586,7 @@ public Task<RedisValue> StreamAddAsync(RedisKey key, NameValueEntry[] streamPair
             return Inner.StreamAddAsync(ToInner(key), streamPairs, messageId, maxLength, useApproximateMaxLength, flags);
         }
 
-        public Task<RedisStreamEntry[]> StreamClaimAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None)
+        public Task<StreamEntry[]> StreamClaimAsync(RedisKey key, RedisValue consumerGroup, RedisValue claimingConsumer, long minIdleTimeInMs, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamClaimAsync(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
         }
@@ -596,9 +596,14 @@ public Task<RedisValue[]> StreamClaimIdsOnlyAsync(RedisKey key, RedisValue consu
             return Inner.StreamClaimIdsOnlyAsync(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
         }
 
-        public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, RedisValue? readFrom = null, CommandFlags flags = CommandFlags.None)
+        public Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamCreateConsumerGroupAsync(ToInner(key), groupName, readFrom, flags);
+            return Inner.StreamConsumerGroupSetPositionAsync(ToInner(key), groupName, position, flags);
+        }
+
+        public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamCreateConsumerGroupAsync(ToInner(key), groupName, position, flags);
         }
 
         public Task<StreamInfo> StreamInfoAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
@@ -626,6 +631,16 @@ public Task<long> StreamDeleteAsync(RedisKey key, RedisValue[] messageIds, Comma
             return Inner.StreamDeleteAsync(ToInner(key), messageIds, flags);
         }
 
+        public Task<long> StreamDeleteConsumerAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamDeleteConsumerAsync(ToInner(key), groupName, consumerName, flags);
+        }
+
+        public Task<bool> StreamDeleteConsumerGroupAsync(RedisKey key, RedisValue groupName, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamDeleteConsumerGroupAsync(ToInner(key), groupName, flags);
+        }
+
         public Task<StreamPendingInfo> StreamPendingAsync(RedisKey key, RedisValue groupName, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamPendingAsync(ToInner(key), groupName, flags);
@@ -636,24 +651,29 @@ public Task<StreamPendingMessageInfo[]> StreamPendingMessagesAsync(RedisKey key,
             return Inner.StreamPendingMessagesAsync(ToInner(key), groupName, count, consumerName, minId, maxId, flags);
         }
 
-        public Task<RedisStreamEntry[]> StreamRangeAsync(RedisKey key, RedisValue? minId = null, RedisValue? maxId = null, int? count = null, Order order = Order.Ascending, CommandFlags flags = CommandFlags.None)
+        public Task<StreamEntry[]> StreamRangeAsync(RedisKey key, RedisValue? minId = null, RedisValue? maxId = null, int? count = null, Order messageOrder = Order.Ascending, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StreamRangeAsync(ToInner(key), minId, maxId, count, messageOrder, flags);
+        }
+
+        public Task<StreamEntry[]> StreamReadAsync(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamRangeAsync(ToInner(key), minId, maxId, count, order, flags);
+            return Inner.StreamReadAsync(ToInner(key), position, count, flags);
         }
 
-        public Task<RedisStreamEntry[]> StreamReadAsync(RedisKey key, RedisValue afterId, int? count = null, CommandFlags flags = CommandFlags.None)
+        public Task<RedisStream[]> StreamReadAsync(StreamPosition[] streamPositions, int? countPerStream = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamReadAsync(ToInner(key), afterId, count, flags);
+            return Inner.StreamReadAsync(streamPositions, countPerStream, flags);
         }
 
-        public Task<RedisStream[]> StreamReadAsync(StreamIdPair[] streamIdPairs, int? countPerStream = null, CommandFlags flags = CommandFlags.None)
+        public Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamReadAsync(streamIdPairs, countPerStream, flags);
+            return Inner.StreamReadGroupAsync(ToInner(key), groupName, consumerName, position, count, flags);
         }
 
-        public Task<RedisStreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? readFromId = null, int? count = null, CommandFlags flags = CommandFlags.None)
+        public Task<RedisStream[]> StreamReadGroupAsync(StreamPosition[] streamPositions, RedisValue groupName, RedisValue consumerName, int? countPerStream = null, CommandFlags flags = CommandFlags.None)
         {
-            return Inner.StreamReadGroupAsync(ToInner(key), groupName, consumerName, readFromId, count, flags);
+            return Inner.StreamReadGroupAsync(streamPositions, groupName, consumerName, countPerStream, flags);
         }
 
         public Task<long> StreamTrimAsync(RedisKey key, int maxLength, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None)

StackExchange.Redis/StackExchange/Redis/Position.cs

+using System;
+
+namespace StackExchange.Redis
+{
+    /// <summary>
+    /// A position within a stream. Defaults to <see cref="Position.New"/>.
+    /// </summary>
+    public struct Position
+    {
+        /// <summary>
+        /// Indicate a position from which to read a stream.
+        /// </summary>
+        /// <param name="readAfter">The position from which to read a stream.</param>
+        public Position(RedisValue readAfter)
+        {
+            if (readAfter == RedisValue.Null) throw new ArgumentNullException(nameof(readAfter), "readAfter cannot be RedisValue.Null.");
+
+            Kind = PositionKind.Explicit;
+            ExplicitValue = readAfter;
+        }
+
+        private Position(PositionKind kind)
+        {
+            Kind = kind;
+            ExplicitValue = RedisValue.Null;
+        }
+
+        private PositionKind Kind { get; }
+
+        private RedisValue ExplicitValue { get; }
+
+        /// <summary>
+        /// Read new messages.
+        /// </summary>
+        public static Position New = new Position(PositionKind.New);
+
+        /// <summary>
+        /// Read from the beginning of a stream.
+        /// </summary>
+        public static Position Beginning = new Position(PositionKind.Beginning);
+
+        internal RedisValue ResolveForCommand(RedisCommand command)
+        {
+            if (Kind == PositionKind.Explicit) return ExplicitValue;
+            if (Kind == PositionKind.Beginning) return StreamConstants.ReadMinValue;
+
+            // PositionKind.New
+            if (command == RedisCommand.XREAD) throw new InvalidOperationException("Position.New cannot be used with StreamRead.");
+            if (command == RedisCommand.XREADGROUP) return StreamConstants.UndeliveredMessages;
+            if (command == RedisCommand.XGROUP) return StreamConstants.NewMessages;
+            
+            throw new ArgumentException($"Unsupported command in ResolveForCommand: {command}.", nameof(command));
+        }
+    }
+}
+
+

StackExchange.Redis/StackExchange/Redis/RedisStream.cs

@@ -5,7 +5,7 @@
     /// </summary>
     public struct RedisStream
     {
-        internal RedisStream(RedisKey key, RedisStreamEntry[] entries)
+        internal RedisStream(RedisKey key, StreamEntry[] entries)
         {
             Key = key;
             Entries = entries;
@@ -19,6 +19,6 @@ internal RedisStream(RedisKey key, RedisStreamEntry[] entries)
         /// <summary>
         /// An arry of entries contained within the stream.
         /// </summary>
-        public RedisStreamEntry[] Entries { get; }
+        public StreamEntry[] Entries { get; }
     }
 }

StackExchange.Redis/StackExchange/Redis/ResultProcessor.cs

@@ -1325,7 +1325,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
             }
         }
 
-        internal sealed class SingleStreamProcessor : StreamProcessorBase<RedisStreamEntry[]>
+        internal sealed class SingleStreamProcessor : StreamProcessorBase<StreamEntry[]>
         {
             private bool skipStreamName;
 
@@ -1339,7 +1339,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
                 if (result.IsNull)
                 {
                     // Server returns 'nil' if no entries are returned for the given stream.
-                    SetResult(message, new RedisStreamEntry[0]);
+                    SetResult(message, new StreamEntry[0]);
                     return true;
                 }
 
@@ -1348,7 +1348,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
                     return false;
                 }
 
-                RedisStreamEntry[] entries = null;
+                StreamEntry[] entries = null;
 
                 if (skipStreamName)
                 {
@@ -1667,7 +1667,7 @@ internal abstract class StreamProcessorBase<T> : ResultProcessor<T>
         {
             // For command response formats see https://redis.io/topics/streams-intro.
 
-            protected RedisStreamEntry[] ParseRedisStreamEntries(RawResult result)
+            protected StreamEntry[] ParseRedisStreamEntries(RawResult result)
             {
                 if (result.Type != ResultType.MultiBulk)
                 {
@@ -1680,7 +1680,7 @@ protected RedisStreamEntry[] ParseRedisStreamEntries(RawResult result)
                 {
                     if (item.IsNull || item.Type != ResultType.MultiBulk)
                     {
-                        return RedisStreamEntry.Null;
+                        return StreamEntry.Null;
                     }
 
                     // Process the Multibulk array for each entry. The entry contains the following elements:
@@ -1688,7 +1688,7 @@ protected RedisStreamEntry[] ParseRedisStreamEntries(RawResult result)
                     //  [1] = Multibulk array of the name/value pairs of the stream entry's data
                     var entryDetails = item.GetItems();
 
-                    return new RedisStreamEntry(id: entryDetails[0].AsRedisValue(),
+                    return new StreamEntry(id: entryDetails[0].AsRedisValue(),
                         values: ParseStreamEntryValues(entryDetails[1]));
                 });
             }

StackExchange.Redis/StackExchange/Redis/StreamConstants.cs

@@ -42,12 +42,18 @@ internal static class StreamConstants
 
         internal static readonly RedisValue Create = "CREATE";
 
+        internal static readonly RedisValue DeleteConsumer = "DELCONSUMER";
+
+        internal static readonly RedisValue Destroy = "DESTROY";
+
         internal static readonly RedisValue Group = "GROUP";
 
         internal static readonly RedisValue Groups = "GROUPS";
 
         internal static readonly RedisValue JustId = "JUSTID";
 
+        internal static readonly RedisValue SetId = "SETID";
+
         internal static readonly RedisValue MaxLen = "MAXLEN";
 
         internal static readonly RedisValue Stream = "STREAM";

StackExchange.Redis/StackExchange/Redis/RedisStreamEntry.cs → StackExchange.Redis/StackExchange/Redis/StreamEntry.cs

@@ -3,9 +3,9 @@
     /// <summary>
     /// Describes an entry contained in a Redis Stream.
     /// </summary>
-    public struct RedisStreamEntry
+    public struct StreamEntry
     {
-        internal RedisStreamEntry(RedisValue id, NameValueEntry[] values)
+        internal StreamEntry(RedisValue id, NameValueEntry[] values)
         {
             Id = id;
             Values = values;
@@ -14,7 +14,7 @@ internal RedisStreamEntry(RedisValue id, NameValueEntry[] values)
         /// <summary>
         /// A null stream entry.
         /// </summary>
-        public static RedisStreamEntry Null { get; } = new RedisStreamEntry(RedisValue.Null, null);
+        public static StreamEntry Null { get; } = new StreamEntry(RedisValue.Null, null);
 
         /// <summary>
         /// The ID assigned to the message.

StackExchange.Redis/StackExchange/Redis/StreamInfo.cs

@@ -10,8 +10,8 @@ public struct StreamInfo
             int radixTreeKeys,
             int radixTreeNodes,
             int groups,
-            RedisStreamEntry firstEntry,
-            RedisStreamEntry lastEntry)
+            StreamEntry firstEntry,
+            StreamEntry lastEntry)
         {
             Length = length;
             RadixTreeKeys = radixTreeKeys;
@@ -44,11 +44,11 @@ public struct StreamInfo
         /// <summary>
         /// The first entry in the stream.
         /// </summary>
-        public RedisStreamEntry FirstEntry { get; }
+        public StreamEntry FirstEntry { get; }
 
         /// <summary>
         /// The last entry in the stream.
         /// </summary>
-        public RedisStreamEntry LastEntry { get; }
+        public StreamEntry LastEntry { get; }
     }
 }

StackExchange.Redis/StackExchange/Redis/StreamIdPair.cs → StackExchange.Redis/StackExchange/Redis/StreamPosition.cs

-
-namespace StackExchange.Redis
+namespace StackExchange.Redis
 {
     /// <summary>
-    /// Describes a pair consisting of the Stream Key and the ID from which to read.
+    /// Describes a pair consisting of the Stream Key and the <see cref="Position"/> from which to begin reading a stream.
     /// </summary>
-    /// <remarks><see cref="IDatabase.StreamRead(StreamIdPair[], int?, CommandFlags)"/></remarks>
-    public struct StreamIdPair
+    public struct StreamPosition
     {
         /// <summary>
-        /// Initializes a <see cref="StreamIdPair"/> value. 
+        /// Initializes a <see cref="StreamPosition"/> value.
         /// </summary>
         /// <param name="key">The key for the stream.</param>
-        /// <param name="id">The ID from which to begin reading the stream.</param>
-        public StreamIdPair(RedisKey key, RedisValue id)
+        /// <param name="position">The position from which to begin reading the stream.</param>
+        public StreamPosition(RedisKey key, Position position)
         {
             Key = key;
-            Id = id;
+            Position = position;
         }
 
         /// <summary>
-        /// The key for the stream.
+        /// The stream key.
         /// </summary>
         public RedisKey Key { get; }
 
         /// <summary>
-        /// The ID from which to begin reading the stream.
+        /// The offset at which to begin reading the stream.
         /// </summary>
-        public RedisValue Id { get; }
-
-        /// <summary>
-        /// See Object.ToString()
-        /// </summary>
-        public override string ToString() => $"{Key}: {Id}";
+        public Position Position { get; }
     }
 }