Skip to content

S
StackExchange.Redis
Commit e8f6b512 authored by Nick Craver's avatar Nick Craver
Browse files
Options

Merge branch 'master' into pipelines

parents 0c2e6ddc ea85ae80
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 978 additions and 180 deletions
StackExchange.Redis.Tests/DatabaseWrapperTests.cs
View file @ e8f6b512
......@@ -838,6 +838,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()
{
......@@ -848,8 +855,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]
......@@ -881,6 +888,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()
{
......@@ -891,37 +912,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/Naming.cs
View file @ e8f6b512
......@@ -88,6 +88,8 @@ public void CheckDatabaseMethodsUseKeys(Type type)
case nameof(IDatabaseAsync.ScriptEvaluateAsync):
case nameof(IDatabase.StreamRead):
case nameof(IDatabase.StreamReadAsync):
case nameof(IDatabase.StreamReadGroup):
case nameof(IDatabase.StreamReadGroupAsync):
continue; // they're fine, but don't want to widen check to return type
}
......@@ -99,7 +101,6 @@ public void CheckDatabaseMethodsUseKeys(Type type)
private static bool UsesKey(Type type)
{
if (type == typeof(RedisKey)) return true;
if (type == typeof(StreamIdPair)) return true;
if (type.IsArray)
{
......
StackExchange.Redis.Tests/Streams.cs
View file @ e8f6b512
This diff is collapsed.
StackExchange.Redis.Tests/WrapperBaseTests.cs
View file @ e8f6b512
......@@ -803,11 +803,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]
......@@ -839,6 +846,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()
{
......@@ -863,23 +884,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 0 → 100644
View file @ e8f6b512
namespace StackExchange.Redis
{
internal enum PositionKind
{
Beginning = 0,
Explicit = 1,
New = 2
}
}
StackExchange.Redis/StackExchange/Redis/KeyspaceIsolation/DatabaseWrapper.cs
View file @ e8f6b512
......@@ -626,7 +626,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);
}
......@@ -636,9 +636,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)
......@@ -666,6 +671,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);
......@@ -676,24 +691,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 messageOrder = 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 RedisStreamEntry[] StreamRead(RedisKey key, RedisValue afterId, int? count = null, CommandFlags flags = CommandFlags.None)
public StreamEntry[] StreamRead(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
{
return Inner.StreamRead(ToInner(key), position, count, flags);
}
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
View file @ e8f6b512
......@@ -606,7 +606,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);
}
......@@ -616,9 +616,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)
......@@ -646,6 +651,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);
......@@ -656,24 +671,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 messageOrder = 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<RedisStreamEntry[]> StreamReadAsync(RedisKey key, RedisValue afterId, int? count = null, CommandFlags flags = CommandFlags.None)
public Task<StreamEntry[]> StreamReadAsync(RedisKey key, Position position, int? count = null, CommandFlags flags = CommandFlags.None)
{
return Inner.StreamReadAsync(ToInner(key), position, count, flags);
}
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 0 → 100644
View file @ e8f6b512
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
View file @ e8f6b512
......@@ -5,7 +5,7 @@
/// </summary>
public readonly 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
View file @ e8f6b512
......@@ -1330,7 +1330,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
}
}
internal sealed class SingleStreamProcessor : StreamProcessorBase<RedisStreamEntry[]>
internal sealed class SingleStreamProcessor : StreamProcessorBase<StreamEntry[]>
{
private readonly bool skipStreamName;
......@@ -1344,7 +1344,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, Array.Empty<RedisStreamEntry>());
SetResult(message, Array.Empty<StreamEntry>());
return true;
}
......@@ -1353,7 +1353,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
return false;
}
RedisStreamEntry[] entries = null;
StreamEntry[] entries = null;
if (skipStreamName)
{
......@@ -1688,7 +1688,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)
{
......@@ -1701,7 +1701,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:
......@@ -1709,7 +1709,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]));
});
}
......@@ -1719,7 +1719,7 @@ protected NameValueEntry[] ParseStreamEntryValues(RawResult result)
// The XRANGE, XREVRANGE, XREAD commands return stream entries
// in the following format. The name/value pairs are interleaved
// in the same fashion as the HGETALL response.
//
//
// 1) 1) 1518951480106-0
// 2) 1) "sensor-id"
// 2) "1234"
......@@ -1817,7 +1817,7 @@ public override bool SetResult(PhysicalConnection connection, Message message, R
}
return final;
}
protected override bool SetResultCore(PhysicalConnection connection, Message message, RawResult result)
{
bool happy;
......
StackExchange.Redis/StackExchange/Redis/StreamConstants.cs
View file @ e8f6b512
......@@ -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
View file @ e8f6b512
......@@ -3,9 +3,9 @@
/// <summary>
/// Describes an entry contained in a Redis Stream.
/// </summary>
public readonly struct RedisStreamEntry
public readonly 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
View file @ e8f6b512
......@@ -10,8 +10,8 @@ namespace StackExchange.Redis
int radixTreeKeys,
int radixTreeNodes,
int groups,
RedisStreamEntry firstEntry,
RedisStreamEntry lastEntry)
StreamEntry firstEntry,
StreamEntry lastEntry)
{
Length = length;
RadixTreeKeys = radixTreeKeys;
......@@ -44,11 +44,11 @@ namespace StackExchange.Redis
/// <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
View file @ e8f6b512

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 readonly 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; }
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment