From 6c364784922d78d5c1c13e757e9cbb3cd4382f2e Mon Sep 17 00:00:00 2001 From: xli Date: Thu, 29 Jun 2023 12:53:54 -0700 Subject: [PATCH 1/5] Support sharded pubsub --- docs/Timeouts.md | 2 +- src/StackExchange.Redis/CommandMap.cs | 4 +-- src/StackExchange.Redis/Enums/RedisCommand.cs | 6 ++++ src/StackExchange.Redis/Message.cs | 4 +++ src/StackExchange.Redis/PhysicalConnection.cs | 23 ++++++++----- .../PublicAPI/PublicAPI.Shipped.txt | 6 ++-- src/StackExchange.Redis/RawResult.cs | 6 ++-- src/StackExchange.Redis/RedisChannel.cs | 33 +++++++++++++++---- src/StackExchange.Redis/RedisSubscriber.cs | 12 ++++--- src/StackExchange.Redis/ResultProcessor.cs | 4 +-- src/StackExchange.Redis/ServerEndPoint.cs | 4 +++ .../RedisRequest.cs | 2 +- 12 files changed, 77 insertions(+), 29 deletions(-) diff --git a/docs/Timeouts.md b/docs/Timeouts.md index 345f25dc5..a40dbe8de 100644 --- a/docs/Timeouts.md +++ b/docs/Timeouts.md @@ -88,7 +88,7 @@ By default Redis Timeout exception(s) includes useful information, which can hel |qs | Queue-Awaiting-Response : {int}|There are x operations currently awaiting replies from redis server.| |aw | Active-Writer: {bool}|| |bw | Backlog-Writer: {enum} | Possible values are Inactive, Started, CheckingForWork, CheckingForTimeout, RecordingTimeout, WritingMessage, Flushing, MarkingInactive, RecordingWriteFailure, RecordingFault, SettingIdle, SpinningDown, Faulted| -|rs | Read-State: {enum}|Possible values are NotStarted, Init, RanToCompletion, Faulted, ReadSync, ReadAsync, UpdateWriteTime, ProcessBuffer, MarkProcessed, TryParseResult, MatchResult, PubSubMessage, PubSubPMessage, Reconfigure, InvokePubSub, DequeueResult, ComputeResult, CompletePendingMessage, NA| +|rs | Read-State: {enum}|Possible values are NotStarted, Init, RanToCompletion, Faulted, ReadSync, ReadAsync, UpdateWriteTime, ProcessBuffer, MarkProcessed, TryParseResult, MatchResult, PubSubMessage, PubSubSMessage, PubSubPMessage, Reconfigure, InvokePubSub, DequeueResult, ComputeResult, CompletePendingMessage, NA| |ws | Write-State: {enum}| Possible values are Initializing, Idle, Writing, Flushing, Flushed, NA| |in | Inbound-Bytes : {long}|there are x bytes waiting to be read from the input stream from redis| |in-pipe | Inbound-Pipe-Bytes: {long}|Bytes waiting to be read| diff --git a/src/StackExchange.Redis/CommandMap.cs b/src/StackExchange.Redis/CommandMap.cs index a12828033..d537a8403 100644 --- a/src/StackExchange.Redis/CommandMap.cs +++ b/src/StackExchange.Redis/CommandMap.cs @@ -31,7 +31,7 @@ public sealed class CommandMap RedisCommand.BLPOP, RedisCommand.BRPOP, RedisCommand.BRPOPLPUSH, // yeah, me neither! - RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, + RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, RedisCommand.SSUBSCRIBE, RedisCommand.SUNSUBSCRIBE, RedisCommand.DISCARD, RedisCommand.EXEC, RedisCommand.MULTI, RedisCommand.UNWATCH, RedisCommand.WATCH, @@ -57,7 +57,7 @@ public sealed class CommandMap RedisCommand.BLPOP, RedisCommand.BRPOP, RedisCommand.BRPOPLPUSH, // yeah, me neither! - RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, + RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, RedisCommand.SSUBSCRIBE, RedisCommand.SUNSUBSCRIBE, RedisCommand.DISCARD, RedisCommand.EXEC, RedisCommand.MULTI, RedisCommand.UNWATCH, RedisCommand.WATCH, diff --git a/src/StackExchange.Redis/Enums/RedisCommand.cs b/src/StackExchange.Redis/Enums/RedisCommand.cs index 40cb5c708..012c8ddae 100644 --- a/src/StackExchange.Redis/Enums/RedisCommand.cs +++ b/src/StackExchange.Redis/Enums/RedisCommand.cs @@ -172,6 +172,7 @@ internal enum RedisCommand SORT, SORT_RO, SPOP, + SPUBLISH, SRANDMEMBER, SREM, STRLEN, @@ -179,6 +180,8 @@ internal enum RedisCommand SUNION, SUNIONSTORE, SSCAN, + SSUBSCRIBE, + SUNSUBSCRIBE, SWAPDB, SYNC, @@ -429,10 +432,13 @@ internal static bool IsPrimaryOnly(this RedisCommand command) case RedisCommand.SMEMBERS: case RedisCommand.SMISMEMBER: case RedisCommand.SORT_RO: + case RedisCommand.SPUBLISH: case RedisCommand.SRANDMEMBER: + case RedisCommand.SSUBSCRIBE: case RedisCommand.STRLEN: case RedisCommand.SUBSCRIBE: case RedisCommand.SUNION: + case RedisCommand.SUNSUBSCRIBE: case RedisCommand.SSCAN: case RedisCommand.SYNC: case RedisCommand.TIME: diff --git a/src/StackExchange.Redis/Message.cs b/src/StackExchange.Redis/Message.cs index a76001756..1c06d5e42 100644 --- a/src/StackExchange.Redis/Message.cs +++ b/src/StackExchange.Redis/Message.cs @@ -500,6 +500,10 @@ internal static bool RequiresDatabase(RedisCommand command) case RedisCommand.SLAVEOF: case RedisCommand.SLOWLOG: case RedisCommand.SUBSCRIBE: + case RedisCommand.SPUBLISH: + case RedisCommand.SSUBSCRIBE: + case RedisCommand.SUBSCRIBE: + case RedisCommand.SUNSUBSCRIBE: case RedisCommand.SWAPDB: case RedisCommand.SYNC: case RedisCommand.TIME: diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index 3c43002a3..b4c667592 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -27,7 +27,7 @@ internal sealed partial class PhysicalConnection : IDisposable private const int DefaultRedisDatabaseCount = 16; - private static readonly CommandBytes message = "message", pmessage = "pmessage"; + private static readonly CommandBytes message = "message", pmessage = "pmessage", smessage = "smessage"; private static readonly Message[] ReusableChangeDatabaseCommands = Enumerable.Range(0, DefaultRedisDatabaseCount).Select( i => Message.Create(i, CommandFlags.FireAndForget, RedisCommand.SELECT)).ToArray(); @@ -1555,9 +1555,9 @@ private void MatchResult(in RawResult result) // out of band message does not match to a queued message var items = result.GetItems(); - if (items.Length >= 3 && items[0].IsEqual(message)) + if (items.Length >= 3 && (items[0].IsEqual(message) || items[0].IsEqual(smessage))) { - _readStatus = ReadStatus.PubSubMessage; + _readStatus = items[0].IsEqual(message) ? ReadStatus.PubSubMessage : ReadStatus.PubSubSMessage; // special-case the configuration change broadcasts (we don't keep that in the usual pub/sub registry) var configChanged = muxer.ConfigurationChangedChannel; @@ -1579,8 +1579,14 @@ private void MatchResult(in RawResult result) } // invoke the handlers - var channel = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal); - Trace("MESSAGE: " + channel); + RedisChannel channel; + if (items[0].IsEqual(message)) { + channel = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal, isSharded: false); + Trace("MESSAGE: " + channel); + } else { + channel = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal, isSharded: true); + Trace("SMESSAGE: " + channel); + } if (!channel.IsNull && TryGetPubSubPayload(items[2], out var payload)) { _readStatus = ReadStatus.InvokePubSub; @@ -1592,18 +1598,18 @@ private void MatchResult(in RawResult result) { _readStatus = ReadStatus.PubSubPMessage; - var channel = items[2].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal); + var channel = items[2].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal, isSharded: false); Trace("PMESSAGE: " + channel); if (!channel.IsNull && TryGetPubSubPayload(items[3], out var payload)) { - var sub = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Pattern); + var sub = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Pattern, isSharded: false); _readStatus = ReadStatus.InvokePubSub; muxer.OnMessage(sub, channel, payload); } return; // AND STOP PROCESSING! } - // if it didn't look like "[p]message", then we still need to process the pending queue + // if it didn't look like "[p|s]message", then we still need to process the pending queue } Trace("Matching result..."); Message? msg; @@ -1936,6 +1942,7 @@ internal enum ReadStatus MatchResult, PubSubMessage, PubSubPMessage, + PubSubSMessage, Reconfigure, InvokePubSub, DequeueResult, diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt index a55d11cc5..78f901854 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt @@ -359,7 +359,7 @@ StackExchange.Redis.ConnectionMultiplexer.PreserveAsyncOrder.set -> void StackExchange.Redis.ConnectionMultiplexer.PublishReconfigure(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long StackExchange.Redis.ConnectionMultiplexer.PublishReconfigureAsync(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! StackExchange.Redis.ConnectionMultiplexer.ReconfigureAsync(string! reason) -> System.Threading.Tasks.Task! -StackExchange.Redis.ConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void +StackExchange.Redis.ConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void StackExchange.Redis.ConnectionMultiplexer.ResetStormLog() -> void StackExchange.Redis.ConnectionMultiplexer.ServerMaintenanceEvent -> System.EventHandler? StackExchange.Redis.ConnectionMultiplexer.StormLogThreshold.get -> int @@ -497,7 +497,7 @@ StackExchange.Redis.IConnectionMultiplexer.PreserveAsyncOrder.get -> bool StackExchange.Redis.IConnectionMultiplexer.PreserveAsyncOrder.set -> void StackExchange.Redis.IConnectionMultiplexer.PublishReconfigure(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long StackExchange.Redis.IConnectionMultiplexer.PublishReconfigureAsync(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! -StackExchange.Redis.IConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void +StackExchange.Redis.IConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void StackExchange.Redis.IConnectionMultiplexer.ResetStormLog() -> void StackExchange.Redis.IConnectionMultiplexer.ServerMaintenanceEvent -> System.EventHandler! StackExchange.Redis.IConnectionMultiplexer.StormLogThreshold.get -> int @@ -1265,7 +1265,9 @@ StackExchange.Redis.RedisChannel.PatternMode.Auto = 0 -> StackExchange.Redis.Red StackExchange.Redis.RedisChannel.PatternMode.Literal = 1 -> StackExchange.Redis.RedisChannel.PatternMode StackExchange.Redis.RedisChannel.PatternMode.Pattern = 2 -> StackExchange.Redis.RedisChannel.PatternMode StackExchange.Redis.RedisChannel.RedisChannel() -> void +StackExchange.Redis.RedisChannel.RedisChannel(byte[]? value, bool isSharded) -> void StackExchange.Redis.RedisChannel.RedisChannel(byte[]? value, StackExchange.Redis.RedisChannel.PatternMode mode) -> void +StackExchange.Redis.RedisChannel.RedisChannel(string! value, bool isSharded) -> void StackExchange.Redis.RedisChannel.RedisChannel(string! value, StackExchange.Redis.RedisChannel.PatternMode mode) -> void StackExchange.Redis.RedisCommandException StackExchange.Redis.RedisCommandException.RedisCommandException(string! message) -> void diff --git a/src/StackExchange.Redis/RawResult.cs b/src/StackExchange.Redis/RawResult.cs index fc189f10c..a0528a5f4 100644 --- a/src/StackExchange.Redis/RawResult.cs +++ b/src/StackExchange.Redis/RawResult.cs @@ -117,7 +117,7 @@ public bool MoveNext() } public ReadOnlySequence Current { get; private set; } } - internal RedisChannel AsRedisChannel(byte[]? channelPrefix, RedisChannel.PatternMode mode) + internal RedisChannel AsRedisChannel(byte[]? channelPrefix, RedisChannel.PatternMode mode, bool isSharded) { switch (Type) { @@ -125,12 +125,12 @@ internal RedisChannel AsRedisChannel(byte[]? channelPrefix, RedisChannel.Pattern case ResultType.BulkString: if (channelPrefix == null) { - return new RedisChannel(GetBlob(), mode); + return isSharded ? new RedisChannel(GetBlob(), true) : new RedisChannel(GetBlob(), mode); } if (StartsWith(channelPrefix)) { byte[] copy = Payload.Slice(channelPrefix.Length).ToArray(); - return new RedisChannel(copy, mode); + return isSharded ? new RedisChannel(copy, true) : new RedisChannel(copy, mode); } return default; default: diff --git a/src/StackExchange.Redis/RedisChannel.cs b/src/StackExchange.Redis/RedisChannel.cs index 5e9446506..2131d5966 100644 --- a/src/StackExchange.Redis/RedisChannel.cs +++ b/src/StackExchange.Redis/RedisChannel.cs @@ -10,6 +10,7 @@ namespace StackExchange.Redis { internal readonly byte[]? Value; internal readonly bool _isPatternBased; + internal readonly bool _isSharded; /// /// Indicates whether the channel-name is either null or a zero-length value. @@ -21,6 +22,11 @@ namespace StackExchange.Redis /// public bool IsPattern => _isPatternBased; + /// + /// Indicates whether this channel represents a shard channel (see SSUBSCRIBE) + /// + public bool IsSharded => _isSharded; + internal bool IsNull => Value == null; @@ -57,7 +63,7 @@ public static bool UseImplicitAutoPattern /// /// The name of the channel to create. /// The mode for name matching. - public RedisChannel(byte[]? value, PatternMode mode) : this(value, DeterminePatternBased(value, mode)) { } + public RedisChannel(byte[]? value, PatternMode mode) : this(value, DeterminePatternBased(value, mode), false) { } /// /// Create a new redis channel from a string, explicitly controlling the pattern mode. @@ -66,10 +72,25 @@ public RedisChannel(byte[]? value, PatternMode mode) : this(value, DeterminePatt /// The mode for name matching. public RedisChannel(string value, PatternMode mode) : this(value == null ? null : Encoding.UTF8.GetBytes(value), mode) { } - private RedisChannel(byte[]? value, bool isPatternBased) + /// + /// Create a new redis channel from a buffer, explicitly controlling the sharding mode. + /// + /// The name of the channel to create. + /// Whether the channel is sharded. + public RedisChannel(byte[]? value, bool isSharded) : this(value, false, isSharded) {} + + /// + /// Create a new redis channel from a string, explicitly controlling the sharding mode. + /// + /// The string name of the channel to create. + /// Whether the channel is sharded. + public RedisChannel(string value, bool isSharded) : this(value == null ? null : Encoding.UTF8.GetBytes(value), isSharded) {} + + private RedisChannel(byte[]? value, bool isPatternBased, bool isSharded) { Value = value; _isPatternBased = isPatternBased; + _isSharded = isSharded; } private static bool DeterminePatternBased(byte[]? value, PatternMode mode) => mode switch @@ -121,7 +142,7 @@ private RedisChannel(byte[]? value, bool isPatternBased) /// The first to compare. /// The second to compare. public static bool operator ==(RedisChannel x, RedisChannel y) => - x._isPatternBased == y._isPatternBased && RedisValue.Equals(x.Value, y.Value); + x._isPatternBased == y._isPatternBased && RedisValue.Equals(x.Value, y.Value) && x._isSharded == y._isSharded; /// /// Indicate whether two channel names are equal. @@ -169,10 +190,10 @@ private RedisChannel(byte[]? value, bool isPatternBased) /// Indicate whether two channel names are equal. /// /// The to compare to. - public bool Equals(RedisChannel other) => _isPatternBased == other._isPatternBased && RedisValue.Equals(Value, other.Value); + public bool Equals(RedisChannel other) => _isPatternBased == other._isPatternBased && RedisValue.Equals(Value, other.Value) && _isSharded == other._isSharded; /// - public override int GetHashCode() => RedisValue.GetHashCode(Value) + (_isPatternBased ? 1 : 0); + public override int GetHashCode() => RedisValue.GetHashCode(Value) + (_isPatternBased ? 1 : 0) + (_isSharded ? 2 : 0); /// /// Obtains a string representation of the channel name. @@ -282,4 +303,4 @@ public static implicit operator RedisChannel(byte[]? key) private RedisChannel(byte[]? value) => throw new NotSupportedException(); #endif } -} +} \ No newline at end of file diff --git a/src/StackExchange.Redis/RedisSubscriber.cs b/src/StackExchange.Redis/RedisSubscriber.cs index 39b99bfe2..066ac2425 100644 --- a/src/StackExchange.Redis/RedisSubscriber.cs +++ b/src/StackExchange.Redis/RedisSubscriber.cs @@ -165,8 +165,11 @@ internal Message GetMessage(RedisChannel channel, SubscriptionAction action, Com SubscriptionAction.Subscribe when isPattern => RedisCommand.PSUBSCRIBE, SubscriptionAction.Unsubscribe when isPattern => RedisCommand.PUNSUBSCRIBE, - SubscriptionAction.Subscribe when !isPattern => RedisCommand.SUBSCRIBE, - SubscriptionAction.Unsubscribe when !isPattern => RedisCommand.UNSUBSCRIBE, + SubscriptionAction.Subscribe when isSharded => RedisCommand.SSUBSCRIBE, + SubscriptionAction.Unsubscribe when isSharded => RedisCommand.SUNSUBSCRIBE, + + SubscriptionAction.Subscribe when !isPattern && !isSharded => RedisCommand.SUBSCRIBE, + SubscriptionAction.Unsubscribe when !isPattern && !isSharded => RedisCommand.UNSUBSCRIBE, _ => throw new ArgumentOutOfRangeException(nameof(action), "This would be an impressive boolean feat"), }; @@ -338,14 +341,14 @@ private static void ThrowIfNull(in RedisChannel channel) public long Publish(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None) { ThrowIfNull(channel); - var msg = Message.Create(-1, flags, RedisCommand.PUBLISH, channel, message); + var msg = channel.IsSharded ? Message.Create(-1, flags, RedisCommand.SPUBLISH, channel, message) : Message.Create(-1, flags, RedisCommand.PUBLISH, channel, message); return ExecuteSync(msg, ResultProcessor.Int64); } public Task PublishAsync(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None) { ThrowIfNull(channel); - var msg = Message.Create(-1, flags, RedisCommand.PUBLISH, channel, message); + var msg = channel.IsSharded ? Message.Create(-1, flags, RedisCommand.SPUBLISH, channel, message) : Message.Create(-1, flags, RedisCommand.PUBLISH, channel, message); return ExecuteAsync(msg, ResultProcessor.Int64); } @@ -485,6 +488,7 @@ private bool UnregisterSubscription(in RedisChannel channel, Action connection.BridgeCouldBeNull?.ServerEndPoint, + RedisCommand.SUBSCRIBE or RedisCommand.SSUBSCRIBE or RedisCommand.PSUBSCRIBE => connection.BridgeCouldBeNull?.ServerEndPoint, _ => null }; Subscription?.SetCurrentServer(newServer); @@ -1325,7 +1325,7 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes { case ResultType.MultiBulk: var final = result.ToArray( - (in RawResult item, in ChannelState state) => item.AsRedisChannel(state.Prefix, state.Mode), + (in RawResult item, in ChannelState state) => item.AsRedisChannel(state.Prefix, state.Mode, isSharded: false), new ChannelState(connection.ChannelPrefix, mode))!; SetResult(message, final); diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index d3082e35c..0b126914c 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -242,6 +242,8 @@ public void Dispose() case RedisCommand.UNSUBSCRIBE: case RedisCommand.PSUBSCRIBE: case RedisCommand.PUNSUBSCRIBE: + case RedisCommand.SSUBSCRIBE: + case RedisCommand.SUNSUBSCRIBE: message.SetForSubscriptionBridge(); break; } @@ -260,6 +262,8 @@ public void Dispose() case RedisCommand.UNSUBSCRIBE: case RedisCommand.PSUBSCRIBE: case RedisCommand.PUNSUBSCRIBE: + case RedisCommand.SSUBSCRIBE: + case RedisCommand.SUNSUBSCRIBE: return subscription ?? (create ? subscription = CreateBridge(ConnectionType.Subscription, null) : null); default: return interactive ?? (create ? interactive = CreateBridge(ConnectionType.Interactive, null) : null); diff --git a/toys/StackExchange.Redis.Server/RedisRequest.cs b/toys/StackExchange.Redis.Server/RedisRequest.cs index 36d73a4bc..4c3d28787 100644 --- a/toys/StackExchange.Redis.Server/RedisRequest.cs +++ b/toys/StackExchange.Redis.Server/RedisRequest.cs @@ -45,7 +45,7 @@ public int GetInt32(int index) public RedisKey GetKey(int index) => _inner[index].AsRedisKey(); public RedisChannel GetChannel(int index, RedisChannel.PatternMode mode) - => _inner[index].AsRedisChannel(null, mode); + => _inner[index].AsRedisChannel(null, mode, false); internal bool TryGetCommandBytes(int i, out CommandBytes command) { From 248c1821506bb4f4483842753927249b7573a936 Mon Sep 17 00:00:00 2001 From: xli Date: Thu, 29 Jun 2023 13:02:49 -0700 Subject: [PATCH 2/5] Support sharded pubsub --- src/StackExchange.Redis/RedisSubscriber.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/StackExchange.Redis/RedisSubscriber.cs b/src/StackExchange.Redis/RedisSubscriber.cs index 066ac2425..52f19e8a6 100644 --- a/src/StackExchange.Redis/RedisSubscriber.cs +++ b/src/StackExchange.Redis/RedisSubscriber.cs @@ -160,6 +160,7 @@ public Subscription(CommandFlags flags) internal Message GetMessage(RedisChannel channel, SubscriptionAction action, CommandFlags flags, bool internalCall) { var isPattern = channel._isPatternBased; + var isSharded = channel._isSharded; var command = action switch { SubscriptionAction.Subscribe when isPattern => RedisCommand.PSUBSCRIBE, From 90e249368ebe6703477aa0cc74e9e1fc28f8cc63 Mon Sep 17 00:00:00 2001 From: xli Date: Thu, 29 Jun 2023 13:06:47 -0700 Subject: [PATCH 3/5] fix api --- src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt index 78f901854..c21fcb0fc 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt @@ -1260,6 +1260,7 @@ StackExchange.Redis.RedisChannel StackExchange.Redis.RedisChannel.Equals(StackExchange.Redis.RedisChannel other) -> bool StackExchange.Redis.RedisChannel.IsNullOrEmpty.get -> bool StackExchange.Redis.RedisChannel.IsPattern.get -> bool +StackExchange.Redis.RedisChannel.IsSharded.get -> bool StackExchange.Redis.RedisChannel.PatternMode StackExchange.Redis.RedisChannel.PatternMode.Auto = 0 -> StackExchange.Redis.RedisChannel.PatternMode StackExchange.Redis.RedisChannel.PatternMode.Literal = 1 -> StackExchange.Redis.RedisChannel.PatternMode From ac44789aebd84f6e59af0eb9bb0dd7170fba8ec3 Mon Sep 17 00:00:00 2001 From: xli Date: Thu, 29 Jun 2023 13:07:54 -0700 Subject: [PATCH 4/5] fix enum --- src/StackExchange.Redis/Message.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/StackExchange.Redis/Message.cs b/src/StackExchange.Redis/Message.cs index 1c06d5e42..bc05b7b55 100644 --- a/src/StackExchange.Redis/Message.cs +++ b/src/StackExchange.Redis/Message.cs @@ -502,7 +502,6 @@ internal static bool RequiresDatabase(RedisCommand command) case RedisCommand.SUBSCRIBE: case RedisCommand.SPUBLISH: case RedisCommand.SSUBSCRIBE: - case RedisCommand.SUBSCRIBE: case RedisCommand.SUNSUBSCRIBE: case RedisCommand.SWAPDB: case RedisCommand.SYNC: From fbc8fd5bbe7b3f41f4174348764742e2344b9f76 Mon Sep 17 00:00:00 2001 From: xli Date: Thu, 29 Jun 2023 13:12:58 -0700 Subject: [PATCH 5/5] fix api --- src/StackExchange.Redis/CommandMap.cs | 4 ++-- src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/StackExchange.Redis/CommandMap.cs b/src/StackExchange.Redis/CommandMap.cs index d537a8403..356e1c2d7 100644 --- a/src/StackExchange.Redis/CommandMap.cs +++ b/src/StackExchange.Redis/CommandMap.cs @@ -31,7 +31,7 @@ public sealed class CommandMap RedisCommand.BLPOP, RedisCommand.BRPOP, RedisCommand.BRPOPLPUSH, // yeah, me neither! - RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, RedisCommand.SSUBSCRIBE, RedisCommand.SUNSUBSCRIBE, + RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, RedisCommand.SPUBLISH, RedisCommand.SSUBSCRIBE, RedisCommand.SUNSUBSCRIBE, RedisCommand.DISCARD, RedisCommand.EXEC, RedisCommand.MULTI, RedisCommand.UNWATCH, RedisCommand.WATCH, @@ -57,7 +57,7 @@ public sealed class CommandMap RedisCommand.BLPOP, RedisCommand.BRPOP, RedisCommand.BRPOPLPUSH, // yeah, me neither! - RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, RedisCommand.SSUBSCRIBE, RedisCommand.SUNSUBSCRIBE, + RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE, RedisCommand.SPUBLISH, RedisCommand.SSUBSCRIBE, RedisCommand.SUNSUBSCRIBE, RedisCommand.DISCARD, RedisCommand.EXEC, RedisCommand.MULTI, RedisCommand.UNWATCH, RedisCommand.WATCH, diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt index c21fcb0fc..260015dc9 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt @@ -359,7 +359,7 @@ StackExchange.Redis.ConnectionMultiplexer.PreserveAsyncOrder.set -> void StackExchange.Redis.ConnectionMultiplexer.PublishReconfigure(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long StackExchange.Redis.ConnectionMultiplexer.PublishReconfigureAsync(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! StackExchange.Redis.ConnectionMultiplexer.ReconfigureAsync(string! reason) -> System.Threading.Tasks.Task! -StackExchange.Redis.ConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void +StackExchange.Redis.ConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void StackExchange.Redis.ConnectionMultiplexer.ResetStormLog() -> void StackExchange.Redis.ConnectionMultiplexer.ServerMaintenanceEvent -> System.EventHandler? StackExchange.Redis.ConnectionMultiplexer.StormLogThreshold.get -> int @@ -497,7 +497,7 @@ StackExchange.Redis.IConnectionMultiplexer.PreserveAsyncOrder.get -> bool StackExchange.Redis.IConnectionMultiplexer.PreserveAsyncOrder.set -> void StackExchange.Redis.IConnectionMultiplexer.PublishReconfigure(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long StackExchange.Redis.IConnectionMultiplexer.PublishReconfigureAsync(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! -StackExchange.Redis.IConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void +StackExchange.Redis.IConnectionMultiplexer.RegisterProfiler(System.Func! profilingSessionProvider) -> void StackExchange.Redis.IConnectionMultiplexer.ResetStormLog() -> void StackExchange.Redis.IConnectionMultiplexer.ServerMaintenanceEvent -> System.EventHandler! StackExchange.Redis.IConnectionMultiplexer.StormLogThreshold.get -> int