Merge pull request #1684 from danielmarbach/disposable

AsyncDisposable
rabbitmq · Sep 19, 2024 · be1d1e3 · be1d1e3
2 parents 13fa6b2 + 68cc34e
commit be1d1e3
Show file tree

Hide file tree

Showing 40 changed files with 1,383 additions and 1,513 deletions.
diff --git a/projects/RabbitMQ.Client.OAuth2/RabbitMQ.Client.OAuth2.csproj b/projects/RabbitMQ.Client.OAuth2/RabbitMQ.Client.OAuth2.csproj
@@ -31,7 +31,7 @@
       https://learn.microsoft.com/en-us/answers/questions/1371494/for-net-standard-2-0-library-why-add-net-core-3-1
       https://devblogs.microsoft.com/dotnet/embracing-nullable-reference-types/#what-should-library-authors-do
     -->
-    <LangVersion>8.0</LangVersion>
+    <LangVersion>9.0</LangVersion>
     <Nullable>enable</Nullable>
   </PropertyGroup>
 

diff --git a/projects/RabbitMQ.Client.OpenTelemetry/RabbitMQ.Client.OpenTelemetry.csproj b/projects/RabbitMQ.Client.OpenTelemetry/RabbitMQ.Client.OpenTelemetry.csproj
@@ -27,7 +27,7 @@
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
     <PackageOutputPath>../../packages</PackageOutputPath>
     <PackageReadmeFile>README.md</PackageReadmeFile>
-    <LangVersion>7.3</LangVersion>
+    <LangVersion>9.0</LangVersion>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)' == 'Release' And '$(CI)' == 'true'">

diff --git a/projects/RabbitMQ.Client/IChannel.cs b/projects/RabbitMQ.Client/IChannel.cs
@@ -42,10 +42,10 @@ namespace RabbitMQ.Client
     /// functionality offered by versions 0-8, 0-8qpid, 0-9 and 0-9-1 of AMQP.
     /// </summary>
     /// <remarks>
-    /// Extends the <see cref="IDisposable"/> interface, so that the "using"
+    /// Extends the <see cref="IDisposable"/> interface and the <see cref="IAsyncDisposable"/> interface, so that the "using"
     /// statement can be used to scope the lifetime of a channel when appropriate.
     /// </remarks>
-    public interface IChannel : IDisposable
+    public interface IChannel : IAsyncDisposable, IDisposable
     {
         /// <summary>
         /// Channel number, unique per connections.

diff --git a/projects/RabbitMQ.Client/IConnection.cs b/projects/RabbitMQ.Client/IConnection.cs
@@ -50,12 +50,12 @@ namespace RabbitMQ.Client
     ///  Alternatively, an API tutorial can be found in the User Guide.
     /// </para>
     /// <para>
-    /// Extends the <see cref="IDisposable"/> interface, so that the "using"
+    /// Extends the <see cref="IDisposable"/> and the <see cref="IAsyncDisposable"/> interface, so that the "using"
     /// statement can be used to scope the lifetime of a connection when
     /// appropriate.
     /// </para>
     /// </remarks>
-    public interface IConnection : INetworkConnection, IDisposable
+    public interface IConnection : INetworkConnection, IAsyncDisposable, IDisposable
     {
         /// <summary>
         /// The maximum channel number this connection supports (0 if unlimited).

diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs
@@ -257,7 +257,9 @@ await _connection.DeleteRecordedChannelAsync(this,
         public override string ToString()
             => InnerChannel.ToString();
 
-        public void Dispose()
+        public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
+
+        public async ValueTask DisposeAsync()
         {
             if (_disposed)
             {
@@ -266,7 +268,8 @@ public void Dispose()
 
             if (IsOpen)
             {
-                this.AbortAsync().GetAwaiter().GetResult();
+                await this.AbortAsync()
+                    .ConfigureAwait(false);
             }
 
             _recordedConsumerTags.Clear();

diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs
@@ -299,11 +299,12 @@ private async ValueTask RecoverExchangesAsync(IConnection connection,
             {
                 try
                 {
-                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
+                    IChannel channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+                    await using (channel.ConfigureAwait(false))
                     {
-                        await recordedExchange.RecoverAsync(ch, cancellationToken)
+                        await recordedExchange.RecoverAsync(channel, cancellationToken)
                             .ConfigureAwait(false);
-                        await ch.CloseAsync(cancellationToken)
+                        await channel.CloseAsync(cancellationToken)
                             .ConfigureAwait(false);
                     }
                 }
@@ -351,11 +352,12 @@ private async Task RecoverQueuesAsync(IConnection connection,
                 try
                 {
                     string newName = string.Empty;
-                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
+                    IChannel channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+                    await using (channel.ConfigureAwait(false))
                     {
-                        newName = await recordedQueue.RecoverAsync(ch, cancellationToken)
+                        newName = await recordedQueue.RecoverAsync(channel, cancellationToken)
                             .ConfigureAwait(false);
-                        await ch.CloseAsync(cancellationToken)
+                        await channel.CloseAsync(cancellationToken)
                             .ConfigureAwait(false);
                     }
                     string oldName = recordedQueue.Name;
@@ -463,11 +465,12 @@ private async ValueTask RecoverBindingsAsync(IConnection connection,
             {
                 try
                 {
-                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
+                    IChannel channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+                    await using (channel.ConfigureAwait(false))
                     {
-                        await binding.RecoverAsync(ch, cancellationToken)
+                        await binding.RecoverAsync(channel, cancellationToken)
                             .ConfigureAwait(false);
-                        await ch.CloseAsync(cancellationToken)
+                        await channel.CloseAsync(cancellationToken)
                             .ConfigureAwait(false);
                     }
                 }

diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.cs
@@ -264,7 +264,9 @@ await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToke
             return channel;
         }
 
-        public void Dispose()
+        public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
+
+        public async ValueTask DisposeAsync()
         {
             if (_disposed)
             {
@@ -273,7 +275,8 @@ public void Dispose()
 
             try
             {
-                _innerConnection.Dispose();
+                await _innerConnection.DisposeAsync()
+                    .ConfigureAwait(false);
             }
             catch (OperationInterruptedException)
             {

diff --git a/projects/RabbitMQ.Client/Impl/ChannelBase.cs b/projects/RabbitMQ.Client/Impl/ChannelBase.cs
@@ -568,6 +568,26 @@ protected virtual void Dispose(bool disposing)
             }
         }
 
+        public async ValueTask DisposeAsync()
+        {
+            await DisposeAsyncCore()
+                .ConfigureAwait(false);
+
+            Dispose(false);
+        }
+
+        protected virtual async ValueTask DisposeAsyncCore()
+        {
+            if (IsOpen)
+            {
+                await this.AbortAsync().ConfigureAwait(false);
+            }
+
+            ConsumerDispatcher.Dispose();
+            _rpcSemaphore.Dispose();
+            _confirmSemaphore?.Dispose();
+        }
+
         public Task ConnectionTuneOkAsync(ushort channelMax, uint frameMax, ushort heartbeat, CancellationToken cancellationToken)
         {
             var method = new ConnectionTuneOk(channelMax, frameMax, heartbeat);

diff --git a/projects/RabbitMQ.Client/Impl/Connection.cs b/projects/RabbitMQ.Client/Impl/Connection.cs
@@ -485,7 +485,9 @@ internal ValueTask WriteAsync(RentedMemory frames, CancellationToken cancellatio
             return _frameHandler.WriteAsync(frames, cancellationToken);
         }
 
-        public void Dispose()
+        public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
+
+        public async ValueTask DisposeAsync()
         {
             if (_disposed)
             {
@@ -496,7 +498,8 @@ public void Dispose()
             {
                 if (IsOpen)
                 {
-                    this.AbortAsync().GetAwaiter().GetResult();
+                    await this.AbortAsync()
+                        .ConfigureAwait(false);
                 }
 
                 _session0.Dispose();

diff --git a/projects/Test/Applications/CreateChannel/CreateChannel.csproj b/projects/Test/Applications/CreateChannel/CreateChannel.csproj
@@ -14,6 +14,7 @@
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
+    <LangVersion>9.0</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>

diff --git a/projects/Test/Applications/CreateChannel/Program.cs b/projects/Test/Applications/CreateChannel/Program.cs
@@ -20,7 +20,7 @@ public static async Task Main()
             doneEvent = new AutoResetEvent(false);
 
             var connectionFactory = new ConnectionFactory { };
-            IConnection connection = await connectionFactory.CreateConnectionAsync();
+            await using IConnection connection = await connectionFactory.CreateConnectionAsync();
 
             var watch = Stopwatch.StartNew();
             _ = Task.Run(async () =>
@@ -55,7 +55,6 @@ public static async Task Main()
             Console.WriteLine();
             Console.WriteLine($"Took {watch.Elapsed.TotalMilliseconds} ms");
 
-            connection.Dispose();
             Console.ReadLine();
         }
     }

diff --git a/projects/Test/Applications/GH-1647/Program.cs b/projects/Test/Applications/GH-1647/Program.cs
@@ -11,12 +11,12 @@
 
 var props = new BasicProperties();
 byte[] msg = Encoding.UTF8.GetBytes("test");
-using var connection = await connectionFactory.CreateConnectionAsync();
+await using var connection = await connectionFactory.CreateConnectionAsync();
 for (int i = 0; i < 300; i++)
 {
     try
     {
-        using var channel = await connection.CreateChannelAsync(); // New channel for each message
+        await using var channel = await connection.CreateChannelAsync(); // New channel for each message
         await Task.Delay(1000);
         await channel.BasicPublishAsync(exchange: string.Empty, routingKey: string.Empty,
             mandatory: false, basicProperties: props, body: msg);

diff --git a/projects/Test/Applications/MassPublish/Program.cs b/projects/Test/Applications/MassPublish/Program.cs
@@ -98,10 +98,10 @@ static Program()
 
         static async Task Main()
         {
-            using IConnection consumeConnection = await s_consumeConnectionFactory.CreateConnectionAsync();
+            await using IConnection consumeConnection = await s_consumeConnectionFactory.CreateConnectionAsync();
             consumeConnection.ConnectionShutdownAsync += ConnectionShutdownAsync;
 
-            using IChannel consumeChannel = await consumeConnection.CreateChannelAsync();
+            await using IChannel consumeChannel = await consumeConnection.CreateChannelAsync();
             consumeChannel.ChannelShutdownAsync += Channel_ChannelShutdownAsync;
             await consumeChannel.BasicQosAsync(prefetchSize: 0, prefetchCount: 128, global: false);
 

diff --git a/projects/Test/Applications/PublisherConfirms/PublisherConfirms.cs b/projects/Test/Applications/PublisherConfirms/PublisherConfirms.cs
@@ -23,8 +23,8 @@ static async Task PublishMessagesIndividuallyAsync()
 {
     Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages individually and handling confirms all at once");
 
-    using IConnection connection = await CreateConnectionAsync();
-    using IChannel channel = await connection.CreateChannelAsync();
+    await using IConnection connection = await CreateConnectionAsync();
+    await using IChannel channel = await connection.CreateChannelAsync();
 
     // declare a server-named queue
     QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
@@ -51,8 +51,8 @@ static async Task PublishMessagesInBatchAsync()
 {
     Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages and handling confirms in batches");
 
-    using IConnection connection = await CreateConnectionAsync();
-    using IChannel channel = await connection.CreateChannelAsync();
+    await using IConnection connection = await CreateConnectionAsync();
+    await using IChannel channel = await connection.CreateChannelAsync();
 
     // declare a server-named queue
     QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
@@ -97,8 +97,8 @@ async Task HandlePublishConfirmsAsynchronously()
 {
     Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages and handling confirms asynchronously");
 
-    using IConnection connection = await CreateConnectionAsync();
-    using IChannel channel = await connection.CreateChannelAsync();
+    await using IConnection connection = await CreateConnectionAsync();
+    await using IChannel channel = await connection.CreateChannelAsync();
 
     // declare a server-named queue
     QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();

diff --git a/projects/Test/Common/Common.csproj b/projects/Test/Common/Common.csproj
@@ -16,7 +16,7 @@
     <AssemblyOriginatorKeyFile>../../rabbit.snk</AssemblyOriginatorKeyFile>
     <SignAssembly>true</SignAssembly>
     <IsTestProject>false</IsTestProject>
-    <LangVersion>7.3</LangVersion>
+    <LangVersion>9.0</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>

diff --git a/projects/Test/Integration/ConnectionRecovery/TestBasicAckAndBasicNack.cs b/projects/Test/Integration/ConnectionRecovery/TestBasicAckAndBasicNack.cs
@@ -52,9 +52,9 @@ public override async Task DisposeAsync()
         {
             ConnectionFactory cf = CreateConnectionFactory();
             cf.ClientProvidedName += "-TearDown";
-            using (IConnection conn = await cf.CreateConnectionAsync())
+            await using (IConnection conn = await cf.CreateConnectionAsync())
             {
-                using (IChannel ch = await conn.CreateChannelAsync())
+                await using (IChannel ch = await conn.CreateChannelAsync())
                 {
                     await ch.QueueDeleteAsync(_queueName);
                     await ch.CloseAsync();

diff --git a/projects/Test/Integration/ConnectionRecovery/TestConnectionRecovery.cs b/projects/Test/Integration/ConnectionRecovery/TestConnectionRecovery.cs
@@ -96,7 +96,7 @@ await _channel.ExchangeDeclareAsync(exchange: exchangeName,
             consumer.ReceivedAsync += MessageReceived;
             await _channel.BasicConsumeAsync(queueName, true, consumer);
 
-            using (IChannel pubCh = await _conn.CreateChannelAsync())
+            await using (IChannel pubCh = await _conn.CreateChannelAsync())
             {
                 await pubCh.BasicPublishAsync(exchange: exchangeName, routingKey: routingKey, body: body);
                 await pubCh.CloseAsync();
@@ -106,7 +106,7 @@ await _channel.ExchangeDeclareAsync(exchange: exchangeName,
 
             await CloseAndWaitForRecoveryAsync();
 
-            using (IChannel pubCh = await _conn.CreateChannelAsync())
+            await using (IChannel pubCh = await _conn.CreateChannelAsync())
             {
                 await pubCh.BasicPublishAsync(exchange: exchangeName, routingKey: "unused", body: body);
                 await pubCh.CloseAsync();

diff --git a/projects/Test/Integration/Integration.csproj b/projects/Test/Integration/Integration.csproj
@@ -16,7 +16,7 @@
     <AssemblyOriginatorKeyFile>../../rabbit.snk</AssemblyOriginatorKeyFile>
     <SignAssembly>true</SignAssembly>
     <IsTestProject>true</IsTestProject>
-    <LangVersion>8.0</LangVersion>
+    <LangVersion>9.0</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>