Fix: Allow reconnect on subscriptions with prefixed redis topic (#8731)

Author: N-Olbert

src/HotChocolate/Core/src/Subscriptions/DefaultPubSub.cs

@@ -170,10 +170,7 @@ private async ValueTask<DefaultTopic<TMessage>> CreateTopicAsync<TMessage>(
     {
         var eventTopic = OnCreateTopic<TMessage>(formattedTopic, bufferCapacity, bufferFullMode);
 
-        eventTopic.Closed += (sender, __) =>
-        {
-            _topics.TryRemove(((DefaultTopic<TMessage>)sender!).Name, out _);
-        };
+        eventTopic.Closed += (_, _) => _topics.TryRemove(formattedTopic, out _);
 
         DiagnosticEvents.Created(formattedTopic);
 

src/HotChocolate/Core/test/Subscriptions.Redis.Tests/RedisIntegrationTests.cs

@@ -63,15 +63,27 @@ public async Task Unsubscribe_Should_RemoveChannel()
 
         await result.DisposeAsync();
 
+        var activeChannelsAfterUnsubscribe =
+            WaitForChannelRemoval(cts, activeChannelsAfterSubscribe.Length, GetActiveChannelsAsync);
+        Assert.True(activeChannelsAfterSubscribe.Length > activeChannelsAfterUnsubscribe);
+    }
+
+    public static int WaitForChannelRemoval(
+        CancellationTokenSource cts,
+        int currentlyActiveChannels,
+        Func<Task<RedisResult[]>> getActiveChannelsAsync)
+    {
+        var activeChannelsAfterUnsubscribe = 0;
         var channelRemovedEvent = new ManualResetEventSlim(false);
 
         _ = Task.Run(async () =>
         {
             while (!cts.Token.IsCancellationRequested)
             {
-                var activeChannels = await GetActiveChannelsAsync();
-                if (activeChannels.Length < activeChannelsAfterSubscribe.Length)
+                var activeChannels = await getActiveChannelsAsync();
+                if (activeChannels.Length < currentlyActiveChannels)
                 {
+                    activeChannelsAfterUnsubscribe = activeChannels.Length;
                     channelRemovedEvent.Set();
                     break;
                 }
@@ -81,6 +93,7 @@ public async Task Unsubscribe_Should_RemoveChannel()
         }, cts.Token);
 
         channelRemovedEvent.Wait(cts.Token);
+        return activeChannelsAfterUnsubscribe;
     }
 
     private async Task<RedisResult[]> GetActiveChannelsAsync()

src/HotChocolate/Core/test/Subscriptions.Redis.Tests/RedisTopicPrefixIntegrationTests.cs

@@ -55,13 +55,44 @@ public async Task Subscribe_Should_Create_Channel_With_Prefix()
             cancellationToken: cts.Token);
 
         var activeChannels = await GetActiveChannelsAsync();
+        Assert.NotEmpty(activeChannels);
+    }
+
+    [Fact]
+    public async Task Unsubscribe_And_Reconnect_Topic()
+    {
+        using var cts = new CancellationTokenSource(Timeout);
+        await using var services = CreateServer<Subscription>();
+
+        var result = await services.ExecuteRequestAsync(
+            "subscription { onMessage }",
+            cancellationToken: cts.Token);
+
+        var activeChannelsAfterSubscribe = await GetActiveChannelsAsync();
+        Assert.NotEmpty(activeChannelsAfterSubscribe);
+
+        await result.DisposeAsync();
+
+        var activeChannelsAfterUnsubscribe = RedisIntegrationTests.WaitForChannelRemoval(
+            cts,
+            activeChannelsAfterSubscribe.Length,
+            GetActiveChannelsAsync);
+        Assert.True(activeChannelsAfterSubscribe.Length > activeChannelsAfterUnsubscribe);
+
+        // reconnect
+        result = await services.ExecuteRequestAsync(
+            "subscription { onMessage }",
+            cancellationToken: cts.Token);
+        activeChannelsAfterSubscribe = await GetActiveChannelsAsync();
+        Assert.True(activeChannelsAfterSubscribe.Length > activeChannelsAfterUnsubscribe);
 
-        Assert.Contains(activeChannels, channel => channel.ToString()!.StartsWith(TopicPrefix));
+        await result.DisposeAsync();
     }
 
     private async Task<RedisResult[]> GetActiveChannelsAsync()
     {
-        return (RedisResult[])(await redisResource.GetConnection().GetDatabase().ExecuteAsync("PUBSUB", "CHANNELS"))!;
+        var channels = (RedisResult[])(await redisResource.GetConnection().GetDatabase().ExecuteAsync("PUBSUB", "CHANNELS"))!;
+        return channels.Where(x => x.ToString()!.StartsWith(TopicPrefix)).ToArray();
     }
 
     protected override void ConfigurePubSub(IRequestExecutorBuilder graphqlBuilder)