Merge branch 'master' into pipelines

StackExchange.Redis.Tests/DatabaseWrapperTests.cs

@@ -841,8 +841,8 @@ public void StreamClaimMessagesReturningIds()
         [Fact]
         public void StreamConsumerGroupSetPosition()
         {
-            wrapper.StreamConsumerGroupSetPosition("key", "group", Position.Beginning, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamConsumerGroupSetPosition("prefix:key", "group", Position.Beginning, CommandFlags.HighPriority));
+            wrapper.StreamConsumerGroupSetPosition("key", "group", StreamPosition.Beginning, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamConsumerGroupSetPosition("prefix:key", "group", StreamPosition.Beginning, CommandFlags.HighPriority));
         }
 
         [Fact]
@@ -855,8 +855,8 @@ public void StreamConsumerInfoGet()
         [Fact]
         public void StreamCreateConsumerGroup()
         {
-            wrapper.StreamCreateConsumerGroup("key", "group", Position.Beginning, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamCreateConsumerGroup("prefix:key", "group", Position.Beginning, CommandFlags.HighPriority));
+            wrapper.StreamCreateConsumerGroup("key", "group", StreamPosition.Beginning, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamCreateConsumerGroup("prefix:key", "group", StreamPosition.Beginning, CommandFlags.HighPriority));
         }
 
         [Fact]
@@ -934,15 +934,15 @@ public void StreamRead_1()
         [Fact]
         public void StreamRead_2()
         {
-            wrapper.StreamRead("key", new Position("0-0"), null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamRead("prefix:key", new Position("0-0"), null, CommandFlags.HighPriority));
+            wrapper.StreamRead("key", "0-0", null, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamRead("prefix:key", "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));
+            wrapper.StreamReadGroup("key", "group", "consumer", "0-0", 10, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadGroup("prefix:key", "group", "consumer", "0-0", 10, CommandFlags.HighPriority));
         }
 
         [Fact]

StackExchange.Redis.Tests/WrapperBaseTests.cs

@@ -806,15 +806,15 @@ public void StreamConsumerInfoGetAsync()
         [Fact]
         public void StreamConsumerGroupSetPositionAsync()
         {
-            wrapper.StreamConsumerGroupSetPositionAsync("key", "group", Position.Beginning, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamConsumerGroupSetPositionAsync("prefix:key", "group", Position.Beginning, CommandFlags.HighPriority));
+            wrapper.StreamConsumerGroupSetPositionAsync("key", "group", StreamPosition.Beginning, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamConsumerGroupSetPositionAsync("prefix:key", "group", StreamPosition.Beginning, CommandFlags.HighPriority));
         }
 
         [Fact]
         public void StreamCreateConsumerGroupAsync()
         {
-            wrapper.StreamCreateConsumerGroupAsync("key", "group", new Position("0-0"), CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamCreateConsumerGroupAsync("prefix:key", "group", new Position("0-0"), CommandFlags.HighPriority));
+            wrapper.StreamCreateConsumerGroupAsync("key", "group", "0-0", CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamCreateConsumerGroupAsync("prefix:key", "group", "0-0", CommandFlags.HighPriority));
         }
 
         [Fact]
@@ -892,15 +892,15 @@ public void StreamReadAsync_1()
         [Fact]
         public void StreamReadAsync_2()
         {
-            wrapper.StreamReadAsync("key", new Position("0-0"), null, CommandFlags.HighPriority);
-            mock.Verify(_ => _.StreamReadAsync("prefix:key", new Position("0-0"), null, CommandFlags.HighPriority));
+            wrapper.StreamReadAsync("key", "0-0", null, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadAsync("prefix:key", "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));
+            wrapper.StreamReadGroupAsync("key", "group", "consumer", StreamPosition.Beginning, 10, CommandFlags.HighPriority);
+            mock.Verify(_ => _.StreamReadGroupAsync("prefix:key", "group", "consumer", StreamPosition.Beginning, 10, CommandFlags.HighPriority));
         }
 
         [Fact]

StackExchange.Redis/StackExchange/Redis/Interfaces/IDatabase.cs

@@ -1524,7 +1524,7 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// <param name="position">The position from which to read for the consumer group.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>True if successful, otherwise false.</returns>
-        bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None);
+        bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, RedisValue position, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Retrieve information about the consumers for the given consumer group. This is the equivalent of calling "XINFO GROUPS key group".
@@ -1541,11 +1541,11 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// </summary>
         /// <param name="key">The key of the stream.</param>
         /// <param name="groupName">The name of the group to create.</param>
-        /// <param name="position">The position to begin reading the stream. Defaults to <see cref="Position.New"/>.</param>
+        /// <param name="position">The position to begin reading the stream. Defaults to <see cref="StreamPosition.NewMessages"/>.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>True if the group was created.</returns>
         /// <remarks>https://redis.io/topics/streams-intro</remarks>
-        bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None);
+        bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, RedisValue? position = null, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Delete messages in the stream. This method does not delete the stream.
@@ -1652,7 +1652,7 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// <returns>Returns a value of <see cref="StreamEntry"/> for each message returned.</returns>
         /// <remarks>Equivalent of calling XREAD COUNT num STREAMS key id.</remarks>
         /// <remarks>https://redis.io/commands/xread</remarks>
-        StreamEntry[] StreamRead(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None);
+        StreamEntry[] StreamRead(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Read from multiple streams.
@@ -1671,12 +1671,12 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// <param name="key">The key of the stream.</param>
         /// <param name="groupName">The name of the consumer group.</param>
         /// <param name="consumerName">The consumer name.</param>
-        /// <param name="position">The position from which to read the stream. Defaults to <see cref="Position.New"/> when null.</param>
+        /// <param name="position">The position from which to read the stream. Defaults to <see cref="StreamPosition.NewMessages"/> when null.</param>
         /// <param name="count">The maximum number of messages to return.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>Returns a value of <see cref="StreamEntry"/> for each message returned.</returns>
         /// <remarks>https://redis.io/commands/xreadgroup</remarks>
-        StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None);
+        StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position = null, int? count = null, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Read from multiple streams into the given consumer group. The consumer group with the given <paramref name="groupName"/>

StackExchange.Redis/StackExchange/Redis/Interfaces/IDatabaseAsync.cs

@@ -1435,7 +1435,7 @@ public interface IDatabaseAsync : IRedisAsync
         /// <param name="position">The position from which to read for the consumer group.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>True if successful, otherwise false.</returns>
-        Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None);
+        Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, RedisValue position, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Retrieve information about the consumers for the given consumer group. This is the equivalent of calling "XINFO GROUPS key group".
@@ -1452,11 +1452,11 @@ public interface IDatabaseAsync : IRedisAsync
         /// </summary>
         /// <param name="key">The key of the stream.</param>
         /// <param name="groupName">The name of the group to create.</param>
-        /// <param name="position">The position to begin reading the stream. Defaults to <see cref="Position.New"/>.</param>
+        /// <param name="position">The position to begin reading the stream. Defaults to <see cref="StreamPosition.NewMessages"/>.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>True if the group was created.</returns>
         /// <remarks>https://redis.io/topics/streams-intro</remarks>
-        Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None);
+        Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, RedisValue? position = null, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Delete messages in the stream. This method does not delete the stream.
@@ -1563,7 +1563,7 @@ public interface IDatabaseAsync : IRedisAsync
         /// <returns>Returns an instance of <see cref="StreamEntry"/> for each message returned.</returns>
         /// <remarks>Equivalent of calling XREAD COUNT num STREAMS key id.</remarks>
         /// <remarks>https://redis.io/commands/xread</remarks>
-        Task<StreamEntry[]> StreamReadAsync(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None);
+        Task<StreamEntry[]> StreamReadAsync(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Read from multiple streams.
@@ -1582,12 +1582,12 @@ public interface IDatabaseAsync : IRedisAsync
         /// <param name="key">The key of the stream.</param>
         /// <param name="groupName">The name of the consumer group.</param>
         /// <param name="consumerName">The consumer name.</param>
-        /// <param name="position">The position from which to read the stream. Defaults to <see cref="Position.New"/> when null.</param>
+        /// <param name="position">The position from which to read the stream. Defaults to <see cref="StreamPosition.NewMessages"/> when null.</param>
         /// <param name="count">The maximum number of messages to return.</param>
         /// <param name="flags">The flags to use for this operation.</param>
         /// <returns>Returns a value of <see cref="StreamEntry"/> for each message returned.</returns>
         /// <remarks>https://redis.io/commands/xreadgroup</remarks>
-        Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None);
+        Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position = null, int? count = null, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Read from multiple streams into the given consumer group. The consumer group with the given <paramref name="groupName"/>

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

@@ -636,12 +636,12 @@ public RedisValue[] StreamClaimIdsOnly(RedisKey key, RedisValue consumerGroup, R
             return Inner.StreamClaimIdsOnly(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
         }
 
-        public bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None)
+        public bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, RedisValue position, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamConsumerGroupSetPosition(ToInner(key), groupName, position, flags);
         }
 
-        public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None)
+        public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, RedisValue? position = null, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamCreateConsumerGroup(ToInner(key), groupName, position, flags);
         }
@@ -696,7 +696,7 @@ public StreamEntry[] StreamRange(RedisKey key, RedisValue? minId = null, RedisVa
             return Inner.StreamRange(ToInner(key), minId, maxId, count, messageOrder, flags);
         }
 
-        public StreamEntry[] StreamRead(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamRead(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamRead(ToInner(key), position, count, flags);
         }
@@ -706,7 +706,7 @@ public RedisStream[] StreamRead(StreamPosition[] streamPositions, int? countPerS
             return Inner.StreamRead(streamPositions, countPerStream, flags);
         }
 
-        public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamReadGroup(ToInner(key), groupName, consumerName, position, count, flags);
         }

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

@@ -616,12 +616,12 @@ public Task<RedisValue[]> StreamClaimIdsOnlyAsync(RedisKey key, RedisValue consu
             return Inner.StreamClaimIdsOnlyAsync(ToInner(key), consumerGroup, claimingConsumer, minIdleTimeInMs, messageIds, flags);
         }
 
-        public Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None)
+        public Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, RedisValue position, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamConsumerGroupSetPositionAsync(ToInner(key), groupName, position, flags);
         }
 
-        public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None)
+        public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, RedisValue? position = null, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamCreateConsumerGroupAsync(ToInner(key), groupName, position, flags);
         }
@@ -676,7 +676,7 @@ public Task<StreamEntry[]> StreamRangeAsync(RedisKey key, RedisValue? minId = nu
             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)
+        public Task<StreamEntry[]> StreamReadAsync(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamReadAsync(ToInner(key), position, count, flags);
         }
@@ -686,7 +686,7 @@ public Task<RedisStream[]> StreamReadAsync(StreamPosition[] streamPositions, int
             return Inner.StreamReadAsync(streamPositions, countPerStream, flags);
         }
 
-        public Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
+        public Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StreamReadGroupAsync(ToInner(key), groupName, consumerName, position, count, flags);
         }

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/RedisDatabase.cs

@@ -1789,7 +1789,7 @@ public Task<RedisValue[]> StreamClaimIdsOnlyAsync(RedisKey key, RedisValue consu
             return ExecuteAsync(msg, ResultProcessor.RedisValueArray);
         }
 
-        public bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None)
+        public bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, RedisValue position, CommandFlags flags = CommandFlags.None)
         {
             var msg = Message.Create(Database,
                 flags,
@@ -1799,13 +1799,13 @@ public bool StreamConsumerGroupSetPosition(RedisKey key, RedisValue groupName, P
                     StreamConstants.SetId,
                     key.AsRedisValue(),
                     groupName,
-                    position.ResolveForCommand(RedisCommand.XGROUP)
+                    StreamPosition.Resolve(position, RedisCommand.XGROUP)
                 });
 
             return ExecuteSync(msg, ResultProcessor.Boolean);
         }
 
-        public Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, Position position, CommandFlags flags = CommandFlags.None)
+        public Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue groupName, RedisValue position, CommandFlags flags = CommandFlags.None)
         {
             var msg = Message.Create(Database,
                 flags,
@@ -1815,15 +1815,15 @@ public Task<bool> StreamConsumerGroupSetPositionAsync(RedisKey key, RedisValue g
                     StreamConstants.SetId,
                     key.AsRedisValue(),
                     groupName,
-                    position.ResolveForCommand(RedisCommand.XGROUP)
+                    StreamPosition.Resolve(position, RedisCommand.XGROUP)
                 });
 
             return ExecuteAsync(msg, ResultProcessor.Boolean);
         }
 
-        public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None)
+        public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, RedisValue? position = null, CommandFlags flags = CommandFlags.None)
         {
-            var actualPosition = position ?? Position.New;
+            var actualPosition = position ?? StreamConstants.NewMessages;
 
             var msg = Message.Create(Database,
                 flags,
@@ -1833,15 +1833,15 @@ public bool StreamCreateConsumerGroup(RedisKey key, RedisValue groupName, Positi
                     StreamConstants.Create,
                     key.AsRedisValue(),
                     groupName,
-                    actualPosition.ResolveForCommand(RedisCommand.XGROUP)
+                    StreamPosition.Resolve(actualPosition, RedisCommand.XGROUP)
                 });
 
             return ExecuteSync(msg, ResultProcessor.Boolean);
         }
 
-        public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, Position? position = null, CommandFlags flags = CommandFlags.None)
+        public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupName, RedisValue? position = null, CommandFlags flags = CommandFlags.None)
         {
-            var actualPosition = position ?? Position.New;
+            var actualPosition = position ?? StreamPosition.NewMessages;
 
             var msg = Message.Create(Database,
                 flags,
@@ -1851,7 +1851,7 @@ public Task<bool> StreamCreateConsumerGroupAsync(RedisKey key, RedisValue groupN
                     StreamConstants.Create,
                     key.AsRedisValue(),
                     groupName,
-                    actualPosition.ResolveForCommand(RedisCommand.XGROUP)
+                    StreamPosition.Resolve(actualPosition, RedisCommand.XGROUP)
                 });
 
             return ExecuteAsync(msg, ResultProcessor.Boolean);
@@ -2069,20 +2069,20 @@ public Task<StreamEntry[]> StreamRangeAsync(RedisKey key, RedisValue? minId = nu
             return ExecuteAsync(msg, ResultProcessor.SingleStream);
         }
 
-        public StreamEntry[] StreamRead(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamRead(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None)
         {
             var msg = GetSingleStreamReadMessage(key,
-                position.ResolveForCommand(RedisCommand.XREAD),
+                StreamPosition.Resolve(position, RedisCommand.XREAD),
                 count,
                 flags);
 
             return ExecuteSync(msg, ResultProcessor.SingleStreamWithNameSkip);
         }
 
-        public Task<StreamEntry[]> StreamReadAsync(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
+        public Task<StreamEntry[]> StreamReadAsync(RedisKey key, RedisValue position, int? count = null, CommandFlags flags = CommandFlags.None)
         {
             var msg = GetSingleStreamReadMessage(key,
-                position.ResolveForCommand(RedisCommand.XREAD),
+                StreamPosition.Resolve(position, RedisCommand.XREAD),
                 count,
                 flags);
 
@@ -2101,28 +2101,28 @@ public Task<RedisStream[]> StreamReadAsync(StreamPosition[] streamPositions, int
             return ExecuteAsync(msg, ResultProcessor.MultiStream);
         }
 
-        public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
+        public StreamEntry[] StreamReadGroup(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
         {
-            var actualPosition = position ?? Position.New;
+            var actualPosition = position ?? StreamPosition.NewMessages;
 
             var msg = GetStreamReadGroupMessage(key,
                 groupName,
                 consumerName,
-                actualPosition.ResolveForCommand(RedisCommand.XREADGROUP),
+                StreamPosition.Resolve(actualPosition, RedisCommand.XREADGROUP),
                 count,
                 flags);
 
             return ExecuteSync(msg, ResultProcessor.SingleStreamWithNameSkip);
         }
 
-        public Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, Position? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
+        public Task<StreamEntry[]> StreamReadGroupAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, RedisValue? position = null, int? count = null, CommandFlags flags = CommandFlags.None)
         {
-            var actualPosition = position ?? Position.New;
+            var actualPosition = position ?? StreamPosition.NewMessages;
 
             var msg = GetStreamReadGroupMessage(key,
                 groupName,
                 consumerName,
-                actualPosition.ResolveForCommand(RedisCommand.XREADGROUP),
+                StreamPosition.Resolve(actualPosition, RedisCommand.XREADGROUP),
                 count,
                 flags);
 
@@ -2545,7 +2545,7 @@ private Message GetMultiStreamReadGroupMessage(StreamPosition[] streamPositions,
             for (var i = 0; i < pairCount; i++)
             {
                 values[offset] = streamPositions[i].Key.AsRedisValue();
-                values[offset + pairCount] = streamPositions[i].Position.ResolveForCommand(RedisCommand.XREADGROUP);
+                values[offset + pairCount] = StreamPosition.Resolve(streamPositions[i].Position, RedisCommand.XREADGROUP);
 
                 offset++;
             }
@@ -2601,7 +2601,7 @@ private Message GetMultiStreamReadMessage(StreamPosition[] streamPositions, int?
             for (var i = 0; i < pairCount; i++)
             {
                 values[offset] = streamPositions[i].Key.AsRedisValue();
-                values[offset + pairCount] = streamPositions[i].Position.ResolveForCommand(RedisCommand.XREAD);
+                values[offset + pairCount] = StreamPosition.Resolve(streamPositions[i].Position, RedisCommand.XREAD);
 
                 offset++;
             }

StackExchange.Redis/StackExchange/Redis/StreamPosition.cs

-namespace StackExchange.Redis
+using System;
+
+namespace StackExchange.Redis
 {
     /// <summary>
     /// Describes a pair consisting of the Stream Key and the <see cref="Position"/> from which to begin reading a stream.
     /// </summary>
     public struct StreamPosition
     {
+        /// <summary>
+        /// Read from the beginning of a stream.
+        /// </summary>
+        public static RedisValue Beginning => StreamConstants.ReadMinValue;
+
+        /// <summary>
+        /// Read new messages.
+        /// </summary>
+        public static RedisValue NewMessages => StreamConstants.NewMessages;
+
         /// <summary>
         /// Initializes a <see cref="StreamPosition"/> value.
         /// </summary>
         /// <param name="key">The key for the stream.</param>
         /// <param name="position">The position from which to begin reading the stream.</param>
-        public StreamPosition(RedisKey key, Position position)
+        public StreamPosition(RedisKey key, RedisValue position)
         {
             Key = key;
             Position = position;
@@ -24,6 +36,22 @@ public StreamPosition(RedisKey key, Position position)
         /// <summary>
         /// The offset at which to begin reading the stream.
         /// </summary>
-        public Position Position { get; }
+        public RedisValue Position { get; }
+
+        internal static RedisValue Resolve(RedisValue value, RedisCommand command)
+        {
+            if (value == NewMessages)
+            {
+                switch (command)
+                {
+                    case RedisCommand.XREAD: throw new InvalidOperationException("StreamPosition.NewMessages cannot be used with StreamRead.");
+                    case RedisCommand.XREADGROUP: return StreamConstants.UndeliveredMessages;
+                    case RedisCommand.XGROUP: return StreamConstants.NewMessages;
+                    default: // new is only valid for the above
+                        throw new ArgumentException($"Unsupported command in StreamPosition.Resolve: {command}.", nameof(command));
+                }
+            }
+            return value;
+        }
     }
 }