Implemented IAsyncDisposable

Tapeti/Config/IMessageContext.cs

@@ -2,11 +2,10 @@
 
 namespace Tapeti.Config
 {
-    /// <inheritdoc />
     /// <summary>
     /// Provides information about the message currently being handled.
     /// </summary>
-    public interface IMessageContext : IDisposable
+    public interface IMessageContext : IAsyncDisposable, IDisposable
     {
         /// <summary>
         /// Provides access to the Tapeti config.
@@ -49,7 +48,7 @@ namespace Tapeti.Config
         /// middleware stages (mostly for IControllerMiddlewareBase descendants).
         /// </summary>
         /// <param name="key">A unique key. It is recommended to prefix it with the package name which hosts the middleware to prevent conflicts</param>
-        /// <param name="value">Will be disposed if the value implements IDisposable</param>
+        /// <param name="value">Will be disposed if the value implements IDisposable or IAsyncDisposable</param>
         void Store(string key, object value);
 
         /// <summary>

Tapeti/Connection/TapetiSubscriber.cs

@@ -25,8 +25,17 @@ namespace Tapeti.Connection
         }
 
 
+        public async ValueTask DisposeAsync()
+        {
+            if (consuming)
+                await Stop();
+        }
+
+
         public void Dispose()
         {
+            if (consuming)
+                Stop().GetAwaiter().GetResult();
         }
 
 

Tapeti/Default/ControllerMessageContext.cs

@@ -1,4 +1,5 @@
-using Tapeti.Config;
+using System.Threading.Tasks;
+using Tapeti.Config;
 
 namespace Tapeti.Default
 {
@@ -41,9 +42,19 @@ namespace Tapeti.Default
         /// <inheritdoc />
         public void Dispose()
         {
+            // Do not call decoratedContext.Dispose - by design
         }
 
 
+        /// <inheritdoc />
+        public ValueTask DisposeAsync()
+        {
+            // Do not call decoratedContext.DisposeAsync - by design
+            return default;
+        }
+
+
+
         /// <inheritdoc />
         public void Store(string key, object value)
         {

Tapeti/Default/MessageContext.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using Tapeti.Config;
 
 namespace Tapeti.Default
@@ -39,6 +40,18 @@ namespace Tapeti.Default
         }
 
 
+        /// <inheritdoc />
+        public async ValueTask DisposeAsync()
+        {
+            foreach (var item in items.Values)
+            {
+                if (item is IAsyncDisposable asyncDisposable)
+                    await asyncDisposable.DisposeAsync();
+            }
+        }
+
+
+
         /// <inheritdoc />
         public void Store(string key, object value)
         {

Tapeti/IConnection.cs

@@ -47,11 +47,10 @@ namespace Tapeti
     public delegate void DisconnectedEventHandler(object sender, DisconnectedEventArgs e);
 
 
-    /// <inheritdoc />
     /// <summary>
     /// Represents a connection to a RabbitMQ server
     /// </summary>
-    public interface IConnection : IDisposable
+    public interface IConnection : IAsyncDisposable, IDisposable
     {
         /// <summary>
         /// Creates a subscriber to consume messages from the bound queues.

Tapeti/ISubscriber.cs

@@ -5,11 +5,10 @@ using System.Threading.Tasks;
 
 namespace Tapeti
 {
-    /// <inheritdoc />
     /// <summary>
     /// Manages subscriptions to queues as configured by the bindings.
     /// </summary>
-    public interface ISubscriber : IDisposable
+    public interface ISubscriber : IAsyncDisposable, IDisposable
     {
         /// <summary>
         /// Starts consuming from the subscribed queues if not already started.

Tapeti/Tapeti.csproj

@@ -18,6 +18,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="5.0.0" />
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
     <PackageReference Include="RabbitMQ.Client" Version="6.2.1" />
     <PackageReference Include="System.Configuration.ConfigurationManager" Version="5.0.0" />

Tapeti/TapetiConnection.cs

@@ -30,6 +30,8 @@ namespace Tapeti
         private readonly Lazy<ITapetiClient> client;
         private TapetiSubscriber subscriber;
 
+        private bool disposed;
+
         /// <summary>
         /// Creates a new instance of a TapetiConnection and registers a default IPublisher
         /// in the IoC container as provided in the config.
@@ -97,12 +99,26 @@ namespace Tapeti
         /// <inheritdoc />
         public void Dispose()
         {
-            Close().Wait();
-
-            subscriber?.Dispose();
+            if (!disposed)
+                DisposeAsync().GetAwaiter().GetResult();
         }
 
 
+        /// <inheritdoc />
+        public async ValueTask DisposeAsync()
+        {
+            if (disposed)
+                return;
+
+            if (subscriber != null)
+                await subscriber.DisposeAsync();
+
+            await Close();
+            disposed = true;
+        }
+
+
+
         private class ConnectionEventListener: IConnectionEventListener
         {
             private readonly TapetiConnection owner;