Gold-plating async tests.

lukebakken · lukebakken · commit 923d31c19f19 · 2023-11-17T10:31:53.000-08:00
diff --git a/projects/Test/AsyncIntegration/AsyncIntegrationFixture.cs b/projects/Test/AsyncIntegration/AsyncIntegrationFixture.cs
@@ -29,6 +29,7 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using RabbitMQ.Client;
 using Xunit;
@@ -53,6 +54,18 @@ protected override void SetUp()
             // InitializeAsync
         }
 
+        protected static Task AssertRanToCompletion(params Task[] tasks)
+        {
+            return AssertRanToCompletion(tasks);
+        }
+
+        protected static async Task AssertRanToCompletion(IEnumerable<Task> tasks)
+        {
+            Task whenAllTask = Task.WhenAll(tasks);
+            await whenAllTask;
+            Assert.Equal(TaskStatus.RanToCompletion, whenAllTask.Status);
+        }
+
         public virtual async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
diff --git a/projects/Test/AsyncIntegration/TestAsyncConsumer.cs b/projects/Test/AsyncIntegration/TestAsyncConsumer.cs
@@ -43,7 +43,8 @@ public class TestAsyncConsumer : AsyncIntegrationFixture
     {
         private readonly ShutdownEventArgs _closeArgs = new ShutdownEventArgs(ShutdownInitiator.Application, Constants.ReplySuccess, "normal shutdown");
 
-        public TestAsyncConsumer(ITestOutputHelper output) : base(output, true, 2)
+        public TestAsyncConsumer(ITestOutputHelper output)
+            : base(output, dispatchConsumersAsync: true, consumerDispatchConcurrency: 2)
         {
         }
 
@@ -113,7 +114,7 @@ public async Task TestBasicRoundtripConcurrent()
             await _channel.BasicConsumeAsync(q.QueueName, true, string.Empty, false, false, null, consumer);
 
             // ensure we get a delivery
-            await Task.WhenAll(publish1SyncSource.Task, publish2SyncSource.Task);
+            await AssertRanToCompletion(publish1SyncSource.Task, publish2SyncSource.Task);
 
             bool result1 = await publish1SyncSource.Task;
             Assert.True(result1, $"1 - Non concurrent dispatch lead to deadlock after {maximumWaitTime}");
@@ -216,7 +217,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                             await _channel.BasicConsumeAsync(queueName, true, string.Empty, false, false, null, consumer);
 
                             // ensure we get a delivery
-                            await Task.WhenAll(publish1SyncSource.Task, publish2SyncSource.Task);
+                            await AssertRanToCompletion(publish1SyncSource.Task, publish2SyncSource.Task);
 
                             bool result1 = await publish1SyncSource.Task;
                             Assert.True(result1, $"Non concurrent dispatch lead to deadlock after {maximumWaitTime}");
@@ -226,7 +227,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                         }
                     });
 
-            await Task.WhenAll(publishTask, consumeTask);
+            await AssertRanToCompletion(publishTask, consumeTask);
         }
 
         [Fact]
diff --git a/projects/Test/AsyncIntegration/TestAsyncConsumerExceptions.cs b/projects/Test/AsyncIntegration/TestAsyncConsumerExceptions.cs
@@ -43,53 +43,54 @@ public class TestAsyncConsumerExceptions : AsyncIntegrationFixture
     {
         private static readonly Exception TestException = new Exception("oops");
 
-        public TestAsyncConsumerExceptions(ITestOutputHelper output) : base(output, true, 1)
+        public TestAsyncConsumerExceptions(ITestOutputHelper output)
+            : base(output, dispatchConsumersAsync: true, consumerDispatchConcurrency: 1)
         {
         }
 
         [Fact]
-        public async Task TestCancelNotificationExceptionHandling()
+        public Task TestCancelNotificationExceptionHandling()
         {
             IBasicConsumer consumer = new ConsumerFailingOnCancel(_channel);
-            await TestExceptionHandlingWith(consumer, async (ch, q, c, ct) =>
+            return TestExceptionHandlingWith(consumer, async (ch, q, c, ct) =>
             {
                 await ch.QueueDeleteAsync(q, false, false);
             });
         }
 
         [Fact]
-        public async Task TestConsumerCancelOkExceptionHandling()
+        public Task TestConsumerCancelOkExceptionHandling()
         {
             IBasicConsumer consumer = new ConsumerFailingOnCancelOk(_channel);
-            await TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.BasicCancelAsync(ct));
+            return TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.BasicCancelAsync(ct));
         }
 
         [Fact]
-        public async Task TestConsumerConsumeOkExceptionHandling()
+        public Task TestConsumerConsumeOkExceptionHandling()
         {
             IBasicConsumer consumer = new ConsumerFailingOnConsumeOk(_channel);
-            await TestExceptionHandlingWith(consumer, async (ch, q, c, ct) => await Task.Yield());
+            return TestExceptionHandlingWith(consumer, async (ch, q, c, ct) => await Task.Yield());
         }
 
         [Fact]
-        public async Task TestConsumerShutdownExceptionHandling()
+        public Task TestConsumerShutdownExceptionHandling()
         {
             IBasicConsumer consumer = new ConsumerFailingOnShutdown(_channel);
-            await TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.CloseAsync());
+            return TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.CloseAsync());
         }
 
         [Fact]
-        public async Task TestDeliveryExceptionHandling()
+        public Task TestDeliveryExceptionHandling()
         {
             IBasicConsumer consumer = new ConsumerFailingOnDelivery(_channel);
-            await TestExceptionHandlingWith(consumer, (ch, q, c, ct) =>
+            return TestExceptionHandlingWith(consumer, (ch, q, c, ct) =>
                 ch.BasicPublishAsync("", q, _encoding.GetBytes("msg")));
         }
 
         protected async Task TestExceptionHandlingWith(IBasicConsumer consumer,
             Func<IChannel, string, IBasicConsumer, string, ValueTask> action)
         {
-            var waitSpan = TimeSpan.FromSeconds(2);
+            var waitSpan = TimeSpan.FromSeconds(5);
             var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
             var cts = new CancellationTokenSource(waitSpan);
             cts.Token.Register(() => tcs.TrySetResult(false));
diff --git a/projects/Test/AsyncIntegration/TestBasicGetAsync.cs b/projects/Test/AsyncIntegration/TestBasicGetAsync.cs
@@ -57,6 +57,8 @@ public async Task TestBasicGet()
 
             QueueDeclareOk queueResultPassive = await _channel.QueueDeclareAsync(queueName, true, true, true, true, null);
             Assert.Equal((uint)0, queueResultPassive.MessageCount);
+
+            Assert.Null(await _channel.BasicGetAsync(queueName, true));
         }
     }
 }
diff --git a/projects/Test/AsyncIntegration/TestBasicPublishAsync.cs b/projects/Test/AsyncIntegration/TestBasicPublishAsync.cs
@@ -29,7 +29,6 @@
 //  Copyright (c) 2011-2020 VMware, Inc. or its affiliates.  All rights reserved.
 //---------------------------------------------------------------------------
 
-using System;
 using System.Threading.Tasks;
 using RabbitMQ.Client;
 using Xunit;
@@ -53,22 +52,20 @@ public async Task TestQueuePurgeAsync()
             await _channel.ConfirmSelectAsync();
 
             QueueDeclareOk q = await _channel.QueueDeclareAsync(string.Empty, false, false, true, false, null);
-            string queueName = q.QueueName;
 
             var publishTask = Task.Run(async () =>
             {
+                byte[] body = GetRandomBody(512);
                 for (int i = 0; i < messageCount; i++)
                 {
-                    byte[] body = _encoding.GetBytes(Guid.NewGuid().ToString());
-                    await _channel.BasicPublishAsync(string.Empty, queueName, body);
+                    await _channel.BasicPublishAsync(string.Empty, q, body);
                 }
+                await _channel.WaitForConfirmsOrDieAsync();
                 publishSyncSource.SetResult(true);
             });
 
-            await _channel.WaitForConfirmsOrDieAsync();
             Assert.True(await publishSyncSource.Task);
-
-            Assert.Equal((uint)messageCount, await _channel.QueuePurgeAsync(queueName));
+            Assert.Equal((uint)messageCount, await _channel.QueuePurgeAsync(q));
         }
     }
 }
diff --git a/projects/Test/AsyncIntegration/TestConcurrentAccessWithSharedConnectionAsync.cs b/projects/Test/AsyncIntegration/TestConcurrentAccessWithSharedConnectionAsync.cs
@@ -151,9 +151,7 @@ private async Task TestConcurrentChannelOperationsAsync(Func<IConnection, Task>
                     tasks.Add(action(_conn));
                 }
             }
-            Task t = Task.WhenAll(tasks);
-            await t;
-            Assert.Equal(TaskStatus.RanToCompletion, t.Status);
+            await AssertRanToCompletion(tasks);
 
             // incorrect frame interleaving in these tests will result
             // in an unrecoverable connection-level exception, thus
diff --git a/projects/Test/AsyncIntegration/TestConfirmSelectAsync.cs b/projects/Test/AsyncIntegration/TestConfirmSelectAsync.cs
@@ -38,6 +38,8 @@ namespace Test.AsyncIntegration
 {
     public class TestConfirmSelectAsync : AsyncIntegrationFixture
     {
+        readonly byte[] _message = GetRandomBody(64);
+
         public TestConfirmSelectAsync(ITestOutputHelper output) : base(output)
         {
         }
@@ -63,7 +65,7 @@ public async Task TestConfirmSelectIdempotency()
 
         private ValueTask Publish()
         {
-            return _channel.BasicPublishAsync("", "amq.fanout", _encoding.GetBytes("message"));
+            return _channel.BasicPublishAsync("", "amq.fanout", _message);
         }
     }
 }
diff --git a/projects/Test/AsyncIntegration/TestExchangeDeclareAsync.cs b/projects/Test/AsyncIntegration/TestExchangeDeclareAsync.cs
@@ -50,7 +50,7 @@ public TestExchangeDeclareAsync(ITestOutputHelper output) : base(output)
         public async Task TestConcurrentExchangeDeclareAndBindAsync()
         {
             var exchangeNames = new ConcurrentBag<string>();
-            var ts = new List<Task>();
+            var tasks = new List<Task>();
             NotSupportedException nse = null;
             for (int i = 0; i < 256; i++)
             {
@@ -70,12 +70,12 @@ async Task f()
                     }
                 }
                 var t = Task.Run(f);
-                ts.Add(t);
+                tasks.Add(t);
             }
 
-            await Task.WhenAll(ts);
+            await AssertRanToCompletion(tasks);
             Assert.Null(nse);
-            ts.Clear();
+            tasks.Clear();
 
             foreach (string exchangeName in exchangeNames)
             {
@@ -93,11 +93,18 @@ async Task f()
                     }
                 }
                 var t = Task.Run(f);
-                ts.Add(t);
+                tasks.Add(t);
             }
 
-            await Task.WhenAll(ts);
+            await AssertRanToCompletion(tasks);
             Assert.Null(nse);
+
+            async Task AssertRanToCompletion(IEnumerable<Task> tasks)
+            {
+                Task whenAllTask = Task.WhenAll(tasks);
+                await whenAllTask;
+                Assert.Equal(TaskStatus.RanToCompletion, whenAllTask.Status);
+            }
         }
     }
 }
diff --git a/projects/Test/AsyncIntegration/TestQueueDeclareAsync.cs b/projects/Test/AsyncIntegration/TestQueueDeclareAsync.cs
@@ -84,8 +84,8 @@ public async void TestConcurrentQueueDeclareAndBindAsync()
                 });
             };
 
-            var ts = new List<Task>();
-            var qs = new ConcurrentBag<string>();
+            var tasks = new List<Task>();
+            var queues = new ConcurrentBag<string>();
 
             NotSupportedException nse = null;
             for (int i = 0; i < 256; i++)
@@ -100,23 +100,23 @@ async Task f()
                         QueueDeclareOk r = await _channel.QueueDeclareAsync(queue: string.Empty, passive: false, false, false, false, null);
                         string queueName = r.QueueName;
                         await _channel.QueueBindAsync(queue: queueName, exchange: "amq.fanout", routingKey: queueName, null);
-                        qs.Add(queueName);
+                        queues.Add(queueName);
                     }
                     catch (NotSupportedException e)
                     {
                         nse = e;
                     }
                 }
                 var t = Task.Run(f);
-                ts.Add(t);
+                tasks.Add(t);
             }
 
-            await Task.WhenAll(ts);
+            await AssertRanToCompletion(tasks);
             Assert.Null(nse);
-            ts.Clear();
+            tasks.Clear();
 
             nse = null;
-            foreach (string q in qs)
+            foreach (string q in queues)
             {
                 async Task f()
                 {
@@ -139,10 +139,10 @@ async Task f()
                     }
                 }
                 var t = Task.Run(f);
-                ts.Add(t);
+                tasks.Add(t);
             }
 
-            await Task.WhenAll(ts);
+            await AssertRanToCompletion(tasks);
             Assert.Null(nse);
             Assert.False(sawShutdown);
         }

Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,8 @@ public class TestAsyncConsumer : AsyncIntegrationFixture`
`43`	`43`	`{`
`44`	`44`	`private readonly ShutdownEventArgs _closeArgs = new ShutdownEventArgs(ShutdownInitiator.Application, Constants.ReplySuccess, "normal shutdown");`
`45`	`45`
`46`		`- public TestAsyncConsumer(ITestOutputHelper output) : base(output, true, 2)`
	`46`	`+ public TestAsyncConsumer(ITestOutputHelper output)`
	`47`	`+ : base(output, dispatchConsumersAsync: true, consumerDispatchConcurrency: 2)`
`47`	`48`	`{`
`48`	`49`	`}`
`49`	`50`
`@@ -113,7 +114,7 @@ public async Task TestBasicRoundtripConcurrent()`
`113`	`114`	`await _channel.BasicConsumeAsync(q.QueueName, true, string.Empty, false, false, null, consumer);`
`114`	`115`
`115`	`116`	`// ensure we get a delivery`
`116`		`- await Task.WhenAll(publish1SyncSource.Task, publish2SyncSource.Task);`
	`117`	`+ await AssertRanToCompletion(publish1SyncSource.Task, publish2SyncSource.Task);`
`117`	`118`
`118`	`119`	`bool result1 = await publish1SyncSource.Task;`
`119`	`120`	`Assert.True(result1, $"1 - Non concurrent dispatch lead to deadlock after {maximumWaitTime}");`
`@@ -216,7 +217,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()`
`216`	`217`	`await _channel.BasicConsumeAsync(queueName, true, string.Empty, false, false, null, consumer);`
`217`	`218`
`218`	`219`	`// ensure we get a delivery`
`219`		`- await Task.WhenAll(publish1SyncSource.Task, publish2SyncSource.Task);`
	`220`	`+ await AssertRanToCompletion(publish1SyncSource.Task, publish2SyncSource.Task);`
`220`	`221`
`221`	`222`	`bool result1 = await publish1SyncSource.Task;`
`222`	`223`	`Assert.True(result1, $"Non concurrent dispatch lead to deadlock after {maximumWaitTime}");`
`@@ -226,7 +227,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()`
`226`	`227`	`}`
`227`	`228`	`});`
`228`	`229`
`229`		`- await Task.WhenAll(publishTask, consumeTask);`
	`230`	`+ await AssertRanToCompletion(publishTask, consumeTask);`
`230`	`231`	`}`
`231`	`232`
`232`	`233`	`[Fact]`
Original file line number	Diff line number	Diff line change
`@@ -43,53 +43,54 @@ public class TestAsyncConsumerExceptions : AsyncIntegrationFixture`
`43`	`43`	`{`
`44`	`44`	`private static readonly Exception TestException = new Exception("oops");`
`45`	`45`
`46`		`- public TestAsyncConsumerExceptions(ITestOutputHelper output) : base(output, true, 1)`
	`46`	`+ public TestAsyncConsumerExceptions(ITestOutputHelper output)`
	`47`	`+ : base(output, dispatchConsumersAsync: true, consumerDispatchConcurrency: 1)`
`47`	`48`	`{`
`48`	`49`	`}`
`49`	`50`
`50`	`51`	`[Fact]`
`51`		`- public async Task TestCancelNotificationExceptionHandling()`
	`52`	`+ public Task TestCancelNotificationExceptionHandling()`
`52`	`53`	`{`
`53`	`54`	`IBasicConsumer consumer = new ConsumerFailingOnCancel(_channel);`
`54`		`- await TestExceptionHandlingWith(consumer, async (ch, q, c, ct) =>`
	`55`	`+ return TestExceptionHandlingWith(consumer, async (ch, q, c, ct) =>`
`55`	`56`	`{`
`56`	`57`	`await ch.QueueDeleteAsync(q, false, false);`
`57`	`58`	`});`
`58`	`59`	`}`
`59`	`60`
`60`	`61`	`[Fact]`
`61`		`- public async Task TestConsumerCancelOkExceptionHandling()`
	`62`	`+ public Task TestConsumerCancelOkExceptionHandling()`
`62`	`63`	`{`
`63`	`64`	`IBasicConsumer consumer = new ConsumerFailingOnCancelOk(_channel);`
`64`		`- await TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.BasicCancelAsync(ct));`
	`65`	`+ return TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.BasicCancelAsync(ct));`
`65`	`66`	`}`
`66`	`67`
`67`	`68`	`[Fact]`
`68`		`- public async Task TestConsumerConsumeOkExceptionHandling()`
	`69`	`+ public Task TestConsumerConsumeOkExceptionHandling()`
`69`	`70`	`{`
`70`	`71`	`IBasicConsumer consumer = new ConsumerFailingOnConsumeOk(_channel);`
`71`		`- await TestExceptionHandlingWith(consumer, async (ch, q, c, ct) => await Task.Yield());`
	`72`	`+ return TestExceptionHandlingWith(consumer, async (ch, q, c, ct) => await Task.Yield());`
`72`	`73`	`}`
`73`	`74`
`74`	`75`	`[Fact]`
`75`		`- public async Task TestConsumerShutdownExceptionHandling()`
	`76`	`+ public Task TestConsumerShutdownExceptionHandling()`
`76`	`77`	`{`
`77`	`78`	`IBasicConsumer consumer = new ConsumerFailingOnShutdown(_channel);`
`78`		`- await TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.CloseAsync());`
	`79`	`+ return TestExceptionHandlingWith(consumer, (ch, q, c, ct) => ch.CloseAsync());`
`79`	`80`	`}`
`80`	`81`
`81`	`82`	`[Fact]`
`82`		`- public async Task TestDeliveryExceptionHandling()`
	`83`	`+ public Task TestDeliveryExceptionHandling()`
`83`	`84`	`{`
`84`	`85`	`IBasicConsumer consumer = new ConsumerFailingOnDelivery(_channel);`
`85`		`- await TestExceptionHandlingWith(consumer, (ch, q, c, ct) =>`
	`86`	`+ return TestExceptionHandlingWith(consumer, (ch, q, c, ct) =>`
`86`	`87`	`ch.BasicPublishAsync("", q, _encoding.GetBytes("msg")));`
`87`	`88`	`}`
`88`	`89`
`89`	`90`	`protected async Task TestExceptionHandlingWith(IBasicConsumer consumer,`
`90`	`91`	`Func<IChannel, string, IBasicConsumer, string, ValueTask> action)`
`91`	`92`	`{`
`92`		`- var waitSpan = TimeSpan.FromSeconds(2);`
	`93`	`+ var waitSpan = TimeSpan.FromSeconds(5);`
`93`	`94`	`var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);`
`94`	`95`	`var cts = new CancellationTokenSource(waitSpan);`
`95`	`96`	`cts.Token.Register(() => tcs.TrySetResult(false));`
Original file line number	Diff line number	Diff line change
`@@ -57,6 +57,8 @@ public async Task TestBasicGet()`
`57`	`57`
`58`	`58`	`QueueDeclareOk queueResultPassive = await _channel.QueueDeclareAsync(queueName, true, true, true, true, null);`
`59`	`59`	`Assert.Equal((uint)0, queueResultPassive.MessageCount);`
	`60`	`+`
	`61`	`+ Assert.Null(await _channel.BasicGetAsync(queueName, true));`
`60`	`62`	`}`
`61`	`63`	`}`
`62`	`64`	`}`
Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,6 @@`
`29`	`29`	`// Copyright (c) 2011-2020 VMware, Inc. or its affiliates. All rights reserved.`
`30`	`30`	`//---------------------------------------------------------------------------`
`31`	`31`
`32`		`-using System;`
`33`	`32`	`using System.Threading.Tasks;`
`34`	`33`	`using RabbitMQ.Client;`
`35`	`34`	`using Xunit;`
`@@ -53,22 +52,20 @@ public async Task TestQueuePurgeAsync()`
`53`	`52`	`await _channel.ConfirmSelectAsync();`
`54`	`53`
`55`	`54`	`QueueDeclareOk q = await _channel.QueueDeclareAsync(string.Empty, false, false, true, false, null);`
`56`		`- string queueName = q.QueueName;`
`57`	`55`
`58`	`56`	`var publishTask = Task.Run(async () =>`
`59`	`57`	`{`
	`58`	`+ byte[] body = GetRandomBody(512);`
`60`	`59`	`for (int i = 0; i < messageCount; i++)`
`61`	`60`	`{`
`62`		`- byte[] body = _encoding.GetBytes(Guid.NewGuid().ToString());`
`63`		`- await _channel.BasicPublishAsync(string.Empty, queueName, body);`
	`61`	`+ await _channel.BasicPublishAsync(string.Empty, q, body);`
`64`	`62`	`}`
	`63`	`+ await _channel.WaitForConfirmsOrDieAsync();`
`65`	`64`	`publishSyncSource.SetResult(true);`
`66`	`65`	`});`
`67`	`66`
`68`		`- await _channel.WaitForConfirmsOrDieAsync();`
`69`	`67`	`Assert.True(await publishSyncSource.Task);`
`70`		`-`
`71`		`- Assert.Equal((uint)messageCount, await _channel.QueuePurgeAsync(queueName));`
	`68`	`+ Assert.Equal((uint)messageCount, await _channel.QueuePurgeAsync(q));`
`72`	`69`	`}`
`73`	`70`	`}`
`74`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -151,9 +151,7 @@ private async Task TestConcurrentChannelOperationsAsync(Func<IConnection, Task>`
`151`	`151`	`tasks.Add(action(_conn));`
`152`	`152`	`}`
`153`	`153`	`}`
`154`		`- Task t = Task.WhenAll(tasks);`
`155`		`- await t;`
`156`		`- Assert.Equal(TaskStatus.RanToCompletion, t.Status);`
	`154`	`+ await AssertRanToCompletion(tasks);`
`157`	`155`
`158`	`156`	`// incorrect frame interleaving in these tests will result`
`159`	`157`	`// in an unrecoverable connection-level exception, thus`
Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,8 @@ namespace Test.AsyncIntegration`
`38`	`38`	`{`
`39`	`39`	`public class TestConfirmSelectAsync : AsyncIntegrationFixture`
`40`	`40`	`{`
	`41`	`+ readonly byte[] _message = GetRandomBody(64);`
	`42`	`+`
`41`	`43`	`public TestConfirmSelectAsync(ITestOutputHelper output) : base(output)`
`42`	`44`	`{`
`43`	`45`	`}`
`@@ -63,7 +65,7 @@ public async Task TestConfirmSelectIdempotency()`
`63`	`65`
`64`	`66`	`private ValueTask Publish()`
`65`	`67`	`{`
`66`		`- return _channel.BasicPublishAsync("", "amq.fanout", _encoding.GetBytes("message"));`
	`68`	`+ return _channel.BasicPublishAsync("", "amq.fanout", _message);`
`67`	`69`	`}`
`68`	`70`	`}`
`69`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ public TestExchangeDeclareAsync(ITestOutputHelper output) : base(output)`
`50`	`50`	`public async Task TestConcurrentExchangeDeclareAndBindAsync()`
`51`	`51`	`{`
`52`	`52`	`var exchangeNames = new ConcurrentBag<string>();`
`53`		`- var ts = new List<Task>();`
	`53`	`+ var tasks = new List<Task>();`
`54`	`54`	`NotSupportedException nse = null;`
`55`	`55`	`for (int i = 0; i < 256; i++)`
`56`	`56`	`{`
`@@ -70,12 +70,12 @@ async Task f()`
`70`	`70`	`}`
`71`	`71`	`}`
`72`	`72`	`var t = Task.Run(f);`
`73`		`- ts.Add(t);`
	`73`	`+ tasks.Add(t);`
`74`	`74`	`}`
`75`	`75`
`76`		`- await Task.WhenAll(ts);`
	`76`	`+ await AssertRanToCompletion(tasks);`
`77`	`77`	`Assert.Null(nse);`
`78`		`- ts.Clear();`
	`78`	`+ tasks.Clear();`
`79`	`79`
`80`	`80`	`foreach (string exchangeName in exchangeNames)`
`81`	`81`	`{`
`@@ -93,11 +93,18 @@ async Task f()`
`93`	`93`	`}`
`94`	`94`	`}`
`95`	`95`	`var t = Task.Run(f);`
`96`		`- ts.Add(t);`
	`96`	`+ tasks.Add(t);`
`97`	`97`	`}`
`98`	`98`
`99`		`- await Task.WhenAll(ts);`
	`99`	`+ await AssertRanToCompletion(tasks);`
`100`	`100`	`Assert.Null(nse);`
	`101`	`+`
	`102`	`+ async Task AssertRanToCompletion(IEnumerable<Task> tasks)`
	`103`	`+ {`
	`104`	`+ Task whenAllTask = Task.WhenAll(tasks);`
	`105`	`+ await whenAllTask;`
	`106`	`+ Assert.Equal(TaskStatus.RanToCompletion, whenAllTask.Status);`
	`107`	`+ }`
`101`	`108`	`}`
`102`	`109`	`}`
`103`	`110`	`}`
Original file line number	Diff line number	Diff line change
`@@ -84,8 +84,8 @@ public async void TestConcurrentQueueDeclareAndBindAsync()`
`84`	`84`	`});`
`85`	`85`	`};`
`86`	`86`
`87`		`- var ts = new List<Task>();`
`88`		`- var qs = new ConcurrentBag<string>();`
	`87`	`+ var tasks = new List<Task>();`
	`88`	`+ var queues = new ConcurrentBag<string>();`
`89`	`89`
`90`	`90`	`NotSupportedException nse = null;`
`91`	`91`	`for (int i = 0; i < 256; i++)`
`@@ -100,23 +100,23 @@ async Task f()`
`100`	`100`	`QueueDeclareOk r = await _channel.QueueDeclareAsync(queue: string.Empty, passive: false, false, false, false, null);`
`101`	`101`	`string queueName = r.QueueName;`
`102`	`102`	`await _channel.QueueBindAsync(queue: queueName, exchange: "amq.fanout", routingKey: queueName, null);`
`103`		`- qs.Add(queueName);`
	`103`	`+ queues.Add(queueName);`
`104`	`104`	`}`
`105`	`105`	`catch (NotSupportedException e)`
`106`	`106`	`{`
`107`	`107`	`nse = e;`
`108`	`108`	`}`
`109`	`109`	`}`
`110`	`110`	`var t = Task.Run(f);`
`111`		`- ts.Add(t);`
	`111`	`+ tasks.Add(t);`
`112`	`112`	`}`
`113`	`113`
`114`		`- await Task.WhenAll(ts);`
	`114`	`+ await AssertRanToCompletion(tasks);`
`115`	`115`	`Assert.Null(nse);`
`116`		`- ts.Clear();`
	`116`	`+ tasks.Clear();`
`117`	`117`
`118`	`118`	`nse = null;`
`119`		`- foreach (string q in qs)`
	`119`	`+ foreach (string q in queues)`
`120`	`120`	`{`
`121`	`121`	`async Task f()`
`122`	`122`	`{`
`@@ -139,10 +139,10 @@ async Task f()`
`139`	`139`	`}`
`140`	`140`	`}`
`141`	`141`	`var t = Task.Run(f);`
`142`		`- ts.Add(t);`
	`142`	`+ tasks.Add(t);`
`143`	`143`	`}`
`144`	`144`
`145`		`- await Task.WhenAll(ts);`
	`145`	`+ await AssertRanToCompletion(tasks);`
`146`	`146`	`Assert.Null(nse);`
`147`	`147`	`Assert.False(sawShutdown);`
`148`	`148`	`}`