Unsubscribe from redis topic async

Author: dheardal

src/HotChocolate/Core/src/Subscriptions.Nats/NatsTopic.cs

@@ -23,7 +23,7 @@ public NatsTopic(
         _serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
     }
 
-    protected override async ValueTask<IDisposable> OnConnectAsync(
+    protected override async ValueTask<IAsyncDisposable> OnConnectAsync(
         CancellationToken cancellationToken)
     {
         // We ensure that the processing is not started before the context is fully initialized.
@@ -39,7 +39,7 @@ protected override async ValueTask<IDisposable> OnConnectAsync(
         return new Session(Name, natsSession, DiagnosticEvents);
     }
 
-    private sealed class Session : IDisposable
+    private sealed class Session : IAsyncDisposable
     {
         private readonly string _name;
         private readonly IDisposable _natsSession;
@@ -56,14 +56,16 @@ public Session(
             _diagnosticEvents = diagnosticEvents;
         }
 
-        public void Dispose()
+        public ValueTask DisposeAsync()
         {
             if (!_disposed)
             {
                 _natsSession.Dispose();
                 _diagnosticEvents.ProviderTopicInfo(_name, Session_Dispose_UnsubscribedFromNats);
                 _disposed = true;
             }
+
+            return ValueTask.CompletedTask;
         }
     }
 }

src/HotChocolate/Core/src/Subscriptions.Postgres/PostgresChannel.cs

@@ -39,7 +39,7 @@ public async ValueTask EnsureInitialized(CancellationToken cancellationToken)
         }
     }
 
-    public IDisposable Subscribe(PostgresChannelObserver observer)
+    public IAsyncDisposable Subscribe(PostgresChannelObserver observer)
     {
         _observers.Add(observer);
 
@@ -124,7 +124,7 @@ private void OnNotification(object sender, NpgsqlNotificationEventArgs eventArgs
         }
     }
 
-    private sealed class Unsubscriber : IDisposable
+    private sealed class Unsubscriber : IAsyncDisposable
     {
         private readonly PostgresChannel _channel;
         private readonly PostgresChannelObserver _observer;
@@ -136,9 +136,10 @@ public Unsubscriber(PostgresChannel channel, PostgresChannelObserver observer)
         }
 
         /// <inheritdoc />
-        public void Dispose()
+        public ValueTask DisposeAsync()
         {
             _channel.Unsubscribe(_observer);
+            return ValueTask.CompletedTask;
         }
     }
 

src/HotChocolate/Core/src/Subscriptions.Postgres/PostgresTopic.cs

@@ -22,7 +22,7 @@ public PostgresTopic(
     }
 
     /// <inheritdoc />
-    protected override async ValueTask<IDisposable> OnConnectAsync(
+    protected override async ValueTask<IAsyncDisposable> OnConnectAsync(
         CancellationToken cancellationToken)
     {
         await _channel.EnsureInitialized(cancellationToken);

src/HotChocolate/Core/src/Subscriptions.RabbitMQ/RabbitMQTopic.cs

@@ -25,7 +25,7 @@ public RabbitMQTopic(
         _serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
     }
 
-    protected override async ValueTask<IDisposable> OnConnectAsync(
+    protected override async ValueTask<IAsyncDisposable> OnConnectAsync(
         CancellationToken cancellationToken)
     {
         // We ensure that the processing is not started before the context is fully initialized.
@@ -85,7 +85,7 @@ private AsyncEventingBasicConsumer CreateConsumer(IModel channel, string queueNa
         return new AsyncEventingBasicConsumer(channel);
     }
 
-    private sealed class Subscription : IDisposable
+    private sealed class Subscription : IAsyncDisposable
     {
         private readonly Action _unsubscribe;
         private bool _disposed;
@@ -95,15 +95,17 @@ public Subscription(Action unsubscribe)
             _unsubscribe = unsubscribe;
         }
 
-        public void Dispose()
+        public ValueTask DisposeAsync()
         {
             if (_disposed)
             {
-                return;
+                return ValueTask.CompletedTask;
             }
 
             _unsubscribe();
             _disposed = true;
+
+            return ValueTask.CompletedTask;
         }
     }
 }

src/HotChocolate/Core/src/Subscriptions.Redis/RedisTopic.cs

@@ -23,7 +23,7 @@ public RedisTopic(
         _serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
     }
 
-    protected override async ValueTask<IDisposable> OnConnectAsync(
+    protected override async ValueTask<IAsyncDisposable> OnConnectAsync(
         CancellationToken cancellationToken)
     {
         // We ensure that the processing is not started before the context is fully initialized.
@@ -37,7 +37,7 @@ protected override async ValueTask<IDisposable> OnConnectAsync(
         return new Session(Name, _connection, DiagnosticEvents);
     }
 
-    private sealed class Session : IDisposable
+    private sealed class Session : IAsyncDisposable
     {
         private readonly string _name;
         private readonly IConnectionMultiplexer _connection;
@@ -54,11 +54,11 @@ public Session(
             _diagnosticEvents = diagnosticEvents;
         }
 
-        public void Dispose()
+        public async ValueTask DisposeAsync()
         {
             if (!_disposed)
             {
-                _connection.GetSubscriber().Unsubscribe(_name);
+                await _connection.GetSubscriber().UnsubscribeAsync(_name).ConfigureAwait(false);
                 _diagnosticEvents.ProviderTopicInfo(_name, RedisTopic_UnsubscribedFromRedis);
                 _disposed = true;
             }

src/HotChocolate/Core/src/Subscriptions/DefaultTopic.cs

@@ -193,10 +193,10 @@ internal async ValueTask ConnectAsync(CancellationToken ct = default)
         }
     }
 
-    private void BeginProcessing(IDisposable session)
+    private void BeginProcessing(IAsyncDisposable session)
         => ProcessMessagesSessionAsync(session).FireAndForget();
 
-    private async Task ProcessMessagesSessionAsync(IDisposable session)
+    private async Task ProcessMessagesSessionAsync(IAsyncDisposable session)
     {
         try
         {
@@ -208,7 +208,7 @@ private async Task ProcessMessagesSessionAsync(IDisposable session)
         }
         finally
         {
-            session.Dispose();
+            await session.DisposeAsync().ConfigureAwait(false);
             DiagnosticEvents.Disconnected(Name);
         }
     }
@@ -372,7 +372,7 @@ private MessageEnvelope<TMessage> DeserializeMessage(IMessageSerializer serializ
     /// <returns>
     /// Returns a session to dispose the subscription session.
     /// </returns>
-    protected virtual ValueTask<IDisposable> OnConnectAsync(
+    protected virtual ValueTask<IAsyncDisposable> OnConnectAsync(
         CancellationToken cancellationToken)
         => new(DefaultSession.Instance);
 
@@ -407,10 +407,11 @@ protected virtual void Dispose(bool disposing)
         }
     }
 
-    private sealed class DefaultSession : IDisposable
+    private sealed class DefaultSession : IAsyncDisposable
     {
         private DefaultSession() { }
-        public void Dispose() { }
+
+        public ValueTask DisposeAsync() => ValueTask.CompletedTask;
 
         public static readonly DefaultSession Instance = new();
     }

src/HotChocolate/Core/test/Subscriptions.Postgres.Tests/PostgresChannelTests.cs

@@ -265,7 +265,7 @@ public async Task Usubscribe_Should_StopListeningToMessages()
         SpinWait.SpinUntil(() => receivedMessages.Count == 1, TimeSpan.FromSeconds(1));
 
         // Act
-        disposable.Dispose();
+        await disposable.DisposeAsync();
         await testChannel.SendMessageAsync("aaaaaaaaaaaaaaaaaaaaaaaa:dGVzdA==:foobar");
 
         // Assert