Pub/Sub: Add Blocking Subscribe/Unsubscribe Commands, Parametrized Tests (#211)

Signed-off-by: currantw <taylor.curran@improving.com>

Author: currantw

rust/src/ffi.rs

@@ -153,7 +153,7 @@ unsafe fn convert_pubsub_config(
         );
     }
 
-    // Convert shard channels
+    // Convert sharded channels
     if config.sharded_channel_count > 0 {
         let sharded = unsafe {
             convert_string_array(config.sharded_channels_ptr, config.sharded_channel_count)

sources/Valkey.Glide/Abstract/Subscriber.cs

@@ -11,6 +11,9 @@ namespace Valkey.Glide;
 /// </summary>
 internal sealed class Subscriber : ISubscriber
 {
+    // Default async timeout for StackExchange.Redis compabitility.
+    private static readonly int DefaultTimeoutMs = 5000;
+
     private readonly ConnectionMultiplexer _multiplexer;
     private readonly Database _client;
 
@@ -140,13 +143,14 @@ public async Task UnsubscribeAllAsync(CommandFlags flags = CommandFlags.None)
         _multiplexer.RemoveAllSubscriptions();
 
         // Send unsubscribe commands for all channel modes.
-        await _client.UnsubscribeLazyAsync();
-        await _client.PUnsubscribeLazyAsync();
+        await _client.UnsubscribeAsync();
+        await _client.PUnsubscribeAsync();
 
         if (_client.IsCluster)
         {
+
             // TODO #205: Refactor to use GlideClusterClient instead of custom command.
-            await _client.Command(Request.CustomCommand(["SUNSUBSCRIBE"]), Route.Random);
+            await _client.Command(Request.CustomCommand(["SUNSUBSCRIBE_BLOCKING", GetTimeoutMs().ToString()]), Route.Random);
         }
     }
 
@@ -198,19 +202,22 @@ private async Task<long> SendPublishCommand(ValkeyChannel channel, ValkeyValue m
     private async Task SendSubscribeCommand(ValkeyChannel channel)
     {
         var channelStr = channel.ToString();
+        var timeout = GetTimeout();
 
         if (channel.IsSharded)
         {
             ThrowIfNotClusterMode();
-            await _client.Command(Request.CustomCommand(["SSUBSCRIBE", channelStr]), Route.Random);
+
+            // TODO #205: Refactor to use GlideClusterClient instead of custom command.
+            await _client.Command(Request.CustomCommand(["SSUBSCRIBE_BLOCKING", channelStr, GetTimeoutMs().ToString()]), Route.Random);
         }
         else if (channel.IsPattern)
         {
-            await _client.PSubscribeLazyAsync(channelStr);
+            await _client.PSubscribeAsync(channelStr, timeout);
         }
         else
         {
-            await _client.SubscribeLazyAsync(channelStr);
+            await _client.SubscribeAsync(channelStr, timeout);
         }
     }
 
@@ -221,24 +228,35 @@ private async Task SendSubscribeCommand(ValkeyChannel channel)
     private async Task SendUnsubscribeCommand(ValkeyChannel channel)
     {
         var channelStr = channel.ToString();
+        var timeout = GetTimeout();
 
         if (channel.IsSharded)
         {
             ThrowIfNotClusterMode();
 
             // TODO #205: Refactor to use GlideClusterClient instead of custom command.
-            await _client.Command(Request.CustomCommand(["SUNSUBSCRIBE", channelStr]), Route.Random);
+            await _client.Command(Request.CustomCommand(["SUNSUBSCRIBE_BLOCKING", channelStr, GetTimeoutMs().ToString()]), Route.Random);
         }
         else if (channel.IsPattern)
         {
-            await _client.PUnsubscribeLazyAsync(channelStr);
+            await _client.PUnsubscribeAsync(channelStr, timeout);
         }
         else
         {
-            await _client.UnsubscribeLazyAsync(channelStr);
+            await _client.UnsubscribeAsync(channelStr, timeout);
         }
     }
 
+    /// <summary>
+    /// Returns the timeout for subscribe/unsubscribe operations.
+    /// </summary>
+    private TimeSpan GetTimeout() => TimeSpan.FromMilliseconds(GetTimeoutMs());
+
+    /// <summary>
+    /// Returns the timeout in milliseconds for subscribe/unsubscribe operations.
+    /// </summary>
+    private int GetTimeoutMs() => _multiplexer.RawConfig.AsyncTimeout ?? DefaultTimeoutMs;
+
     /// <summary>
     /// Throws if the client is not in cluster mode.
     /// </summary>

sources/Valkey.Glide/BaseClient.PubSubCommands.cs

@@ -23,29 +23,89 @@ public async Task<long> PublishAsync(string channel, string message)
     #endregion
     #region SubscribeCommands
 
+    public async Task SubscribeAsync(string channel, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.SubscribeBlocking([channel], (uint)timeout.TotalMilliseconds));
+    }
+
+    public async Task SubscribeAsync(IEnumerable<string> channels, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.SubscribeBlocking(channels.ToGlideStrings(), (uint)timeout.TotalMilliseconds));
+    }
+
     public async Task SubscribeLazyAsync(string channel)
         => await Command(Request.Subscribe([channel]));
 
     public async Task SubscribeLazyAsync(IEnumerable<string> channels)
-        => await Command(Request.Subscribe(channels.ToHashSet().ToGlideStrings()));
+        => await Command(Request.Subscribe(channels.ToGlideStrings()));
+
+    public async Task PSubscribeAsync(string pattern, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.PSubscribeBlocking([pattern], (uint)timeout.TotalMilliseconds));
+    }
+
+    public async Task PSubscribeAsync(IEnumerable<string> patterns, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.PSubscribeBlocking(patterns.ToGlideStrings(), (uint)timeout.TotalMilliseconds));
+    }
 
     public async Task PSubscribeLazyAsync(string pattern)
         => await Command(Request.PSubscribe([pattern]));
 
     public async Task PSubscribeLazyAsync(IEnumerable<string> patterns)
-        => await Command(Request.PSubscribe(patterns.ToHashSet().ToGlideStrings()));
+        => await Command(Request.PSubscribe(patterns.ToGlideStrings()));
 
     #endregion
     #region UnsubscribeCommands
 
+    public async Task UnsubscribeAsync(TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.UnsubscribeBlocking([], (uint)timeout.TotalMilliseconds));
+    }
+
+    public async Task UnsubscribeAsync(string channel, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.UnsubscribeBlocking([channel], (uint)timeout.TotalMilliseconds));
+    }
+
+    public async Task UnsubscribeAsync(IEnumerable<string> channels, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.UnsubscribeBlocking(channels.ToGlideStrings(), (uint)timeout.TotalMilliseconds));
+    }
+
     public async Task UnsubscribeLazyAsync()
         => await Command(Request.Unsubscribe([]));
 
     public async Task UnsubscribeLazyAsync(string channel)
         => await Command(Request.Unsubscribe([channel]));
 
     public async Task UnsubscribeLazyAsync(IEnumerable<string> channels)
-        => await Command(Request.Unsubscribe(channels.ToHashSet().ToGlideStrings()));
+        => await Command(Request.Unsubscribe(channels.ToGlideStrings()));
+
+    public async Task PUnsubscribeAsync(TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.PUnsubscribeBlocking([], (uint)timeout.TotalMilliseconds));
+    }
+
+    public async Task PUnsubscribeAsync(string pattern, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.PUnsubscribeBlocking([pattern], (uint)timeout.TotalMilliseconds));
+    }
+
+    public async Task PUnsubscribeAsync(IEnumerable<string> patterns, TimeSpan timeout = default)
+    {
+        GuardClauses.ThrowIfTimeSpanNegative(timeout);
+        await Command(Request.PUnsubscribeBlocking(patterns.ToGlideStrings(), (uint)timeout.TotalMilliseconds));
+    }
 
     public async Task PUnsubscribeLazyAsync()
         => await Command(Request.PUnsubscribe([]));
@@ -54,7 +114,7 @@ public async Task PUnsubscribeLazyAsync(string pattern)
         => await Command(Request.PUnsubscribe([pattern]));
 
     public async Task PUnsubscribeLazyAsync(IEnumerable<string> patterns)
-        => await Command(Request.PUnsubscribe(patterns.ToHashSet().ToGlideStrings()));
+        => await Command(Request.PUnsubscribe(patterns.ToGlideStrings()));
 
     #endregion
     #region IntrospectionCommands
@@ -66,7 +126,7 @@ public async Task<ISet<string>> PubSubChannelsAsync(string pattern)
         => await Command(Request.PubSubChannels(pattern));
 
     public async Task<Dictionary<string, long>> PubSubNumSubAsync(IEnumerable<string> channels)
-        => await Command(Request.PubSubNumSub(channels.ToHashSet().ToGlideStrings()));
+        => await Command(Request.PubSubNumSub(channels.ToGlideStrings()));
 
     public async Task<long> PubSubNumPatAsync()
         => await Command(Request.PubSubNumPat());

sources/Valkey.Glide/BaseClient.cs

@@ -398,7 +398,7 @@ private static PubSubMessage MarshalPubSubMessage(
         else if (pushKind == PushKind.PushPMessage)
             return PubSubMessage.FromPattern(message, channel, pattern!);
         else if (pushKind == PushKind.PushSMessage)
-            return PubSubMessage.FromShardChannel(message, channel);
+            return PubSubMessage.FromShardedChannel(message, channel);
         else
             throw new ArgumentOutOfRangeException(nameof(pushKind), $"Unsupported PushKind: {pushKind}");
     }