fix toy server pub/sub (always respond)

Author: mgravell

tests/StackExchange.Redis.Tests/PubSubTests.cs

@@ -160,6 +160,8 @@ public async Task Ping()
         await using var conn = ConnectFactory(shared: false);
         var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
+
+        await PingAsync(pub, sub, 5).ForAwait();
         await sub.SubscribeAsync(RedisChannel.Literal(Me()), (_, __) => { }); // to ensure we're in subscriber mode
         await PingAsync(pub, sub, 5).ForAwait();
     }
@@ -232,10 +234,28 @@ private async Task PingAsync(IServer pub, ISubscriber sub, int times = 1)
             // way to prove that is to use TPL objects
             var subTask = sub.PingAsync();
             var pubTask = pub.PingAsync();
-            await Task.WhenAll(subTask, pubTask).ForAwait();
+            try
+            {
+                await Task.WhenAll(subTask, pubTask).ForAwait();
+            }
+            catch (TimeoutException ex)
+            {
+                throw new TimeoutException($"Timeout; sub: {GetState(subTask)}, pub: {GetState(pubTask)}", ex);
+            }
+
+            Log($"sub: {GetState(subTask)}, pub: {GetState(pubTask)}");
 
-            Log($"Sub PING time: {subTask.Result.TotalMilliseconds} ms");
-            Log($"Pub PING time: {pubTask.Result.TotalMilliseconds} ms");
+            static string GetState(Task<TimeSpan> pending)
+            {
+                var status = pending.Status;
+                return status switch
+                {
+                    TaskStatus.RanToCompletion => $"{status} in {pending.Result.TotalMilliseconds:###,##0.0}ms)",
+                    TaskStatus.Faulted when pending.Exception is { InnerExceptions.Count:1 } ae => $"{status}: {ae.InnerExceptions[0].Message}",
+                    TaskStatus.Faulted => $"{status}: {pending.Exception?.Message}",
+                    _ => status.ToString(),
+                };
+            }
         }
     }
 

toys/StackExchange.Redis.Server/RedisServer.PubSub.cs

@@ -231,18 +231,32 @@ private void SendSubUnsubMessage(string kind, RedisChannel channel, int count)
         }
     }
 
+    private ref int GetCountField(RedisChannel channel)
+        => ref channel.IsSharded ? ref shardedCount
+        : ref channel.IsPattern ? ref patternCount
+        : ref simpleCount;
+
     internal void Subscribe(RedisChannel channel)
     {
         Regex glob = channel.IsPattern ? BuildGlob(channel) : null;
         var subs = Subscriptions;
         int count;
+        ref int field = ref GetCountField(channel);
         lock (subs)
         {
-            if (subs.ContainsKey(channel)) return;
-            subs.Add(channel, glob);
-            count = channel.IsSharded ? ++shardedCount
-                : channel.IsPattern ? ++patternCount
-                : ++simpleCount;
+            #if NET
+            count = subs.TryAdd(channel, glob) ? ++field : field;
+            #else
+            if (subs.ContainsKey(channel))
+            {
+                count = field;
+            }
+            else
+            {
+                subs.Add(channel, glob);
+                count = ++field;
+            }
+            #endif
         }
         SendSubUnsubMessage(
             channel.IsSharded ? "ssubscribe"
@@ -271,12 +285,10 @@ internal void Unsubscribe(RedisChannel channel)
         var subs = SubscriptionsIfAny;
         if (subs is null) return;
         int count;
+        ref int field = ref GetCountField(channel);
         lock (subs)
         {
-            if (!subs.Remove(channel)) return;
-            count = channel.IsSharded ? --shardedCount
-                : channel.IsPattern ? --patternCount
-                : --simpleCount;
+            count = subs.Remove(channel) ? --field : field;
         }
         SendSubUnsubMessage(
             channel.IsSharded ? "sunsubscribe"