Use CreateConnectionAsync in AsyncIntegration

lukebakken · lukebakken · commit 9bb4995e60df · 2023-11-10T08:09:41.000-08:00
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs
@@ -80,12 +80,14 @@ internal AutorecoveringConnection(ConnectionConfig config, IEndpointResolver end
 
         internal IConnection Open()
         {
-            return InnerConnection.Open();
+            InnerConnection.Open();
+            return this;
         }
 
-        internal ValueTask<IConnection> OpenAsync()
+        internal async ValueTask<IConnection> OpenAsync()
         {
-            return InnerConnection.OpenAsync();
+            await InnerConnection.OpenAsync();
+            return this;
         }
 
         public event EventHandler<EventArgs> RecoverySucceeded
diff --git a/projects/Test/AsyncIntegration/TestAsyncConsumer.cs b/projects/Test/AsyncIntegration/TestAsyncConsumer.cs
@@ -56,7 +56,8 @@ public async Task InitializeAsync()
             _connFactory = CreateConnectionFactory();
             _connFactory.DispatchConsumersAsync = true;
             _connFactory.ConsumerDispatchConcurrency = 2;
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
@@ -254,7 +255,7 @@ public async Task TestBasicRejectAsync()
             var cf = CreateConnectionFactory();
             cf.DispatchConsumersAsync = true;
 
-            using IConnection connection = cf.CreateConnection();
+            using IConnection connection = await cf.CreateConnectionAsync();
             using IChannel channel = await connection.CreateChannelAsync();
 
             connection.ConnectionShutdown += (o, ea) =>
@@ -337,7 +338,7 @@ public async Task TestBasicAckAsync()
             var cf = CreateConnectionFactory();
             cf.DispatchConsumersAsync = true;
 
-            using IConnection connection = cf.CreateConnection();
+            using IConnection connection = await cf.CreateConnectionAsync();
             using IChannel channel = await connection.CreateChannelAsync();
 
             connection.ConnectionShutdown += (o, ea) =>
@@ -405,7 +406,7 @@ public async Task TestBasicNackAsync()
             var cf = CreateConnectionFactory();
             cf.DispatchConsumersAsync = true;
 
-            using IConnection connection = cf.CreateConnection();
+            using IConnection connection = await cf.CreateConnectionAsync();
             using IChannel channel = await connection.CreateChannelAsync();
 
             connection.ConnectionShutdown += (o, ea) =>
diff --git a/projects/Test/AsyncIntegration/TestAsyncConsumerExceptions.cs b/projects/Test/AsyncIntegration/TestAsyncConsumerExceptions.cs
@@ -56,7 +56,8 @@ public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
             _connFactory.DispatchConsumersAsync = true;
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
@@ -118,22 +119,23 @@ await TestExceptionHandlingWith(consumer, (ch, q, c, ct) =>
         protected async Task TestExceptionHandlingWith(IBasicConsumer consumer,
             Func<IChannel, string, IBasicConsumer, string, ValueTask> action)
         {
-            var semaphore = new SemaphoreSlim(0, 1);
-            bool notified = false;
+            var waitSpan = TimeSpan.FromSeconds(2);
+            var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var cts = new CancellationTokenSource(waitSpan);
+            cts.Token.Register(() => tcs.TrySetResult(false));
+
             string q = await _channel.QueueDeclareAsync(string.Empty, false, false, true, false, null);
             _channel.CallbackException += (ch, evt) =>
             {
                 if (evt.Exception == TestException)
                 {
-                    notified = true;
-                    semaphore.Release();
+                    tcs.SetResult(true);
                 }
             };
 
             string tag = await _channel.BasicConsumeAsync(q, true, string.Empty, false, false, null, consumer);
             await action(_channel, q, consumer, tag);
-            Assert.True(await semaphore.WaitAsync(TimeSpan.FromSeconds(2)));
-            Assert.True(notified);
+            Assert.True(await tcs.Task);
         }
 
         private class ConsumerFailingOnDelivery : AsyncEventingBasicConsumer
diff --git a/projects/Test/AsyncIntegration/TestBasicGetAsync.cs b/projects/Test/AsyncIntegration/TestBasicGetAsync.cs
@@ -50,7 +50,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestBasicPublishAsync.cs b/projects/Test/AsyncIntegration/TestBasicPublishAsync.cs
@@ -51,7 +51,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestConcurrentAccessWithSharedConnectionAsync.cs b/projects/Test/AsyncIntegration/TestConcurrentAccessWithSharedConnectionAsync.cs
@@ -40,7 +40,7 @@
 
 namespace Test.AsyncIntegration
 {
-    public class TestConcurrentAccessWithSharedConnectionAsync : IntegrationFixture
+    public class TestConcurrentAccessWithSharedConnectionAsync : IntegrationFixture, IAsyncLifetime
     {
         private const ushort _messageCount = 200;
 
@@ -49,14 +49,24 @@ public TestConcurrentAccessWithSharedConnectionAsync(ITestOutputHelper output) :
         }
 
         protected override void SetUp()
+        {
+        }
+
+        public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _conn.ConnectionShutdown += HandleConnectionShutdown;
             // NB: not creating _channel because this test suite doesn't use it.
             Assert.Null(_channel);
         }
 
+        public Task DisposeAsync()
+        {
+            return Task.CompletedTask;
+        }
+
         [Fact]
         public Task TestConcurrentChannelOpenAndPublishingWithBlankMessagesAsync()
         {
diff --git a/projects/Test/AsyncIntegration/TestConfirmSelectAsync.cs b/projects/Test/AsyncIntegration/TestConfirmSelectAsync.cs
@@ -50,7 +50,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestExchangeDeclareAsync.cs b/projects/Test/AsyncIntegration/TestExchangeDeclareAsync.cs
@@ -55,7 +55,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestExtensionsAsync.cs b/projects/Test/AsyncIntegration/TestExtensionsAsync.cs
@@ -51,7 +51,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestFloodPublishingAsync.cs b/projects/Test/AsyncIntegration/TestFloodPublishingAsync.cs
@@ -60,7 +60,8 @@ public async Task TestUnthrottledFloodPublishingAsync()
             _connFactory = CreateConnectionFactory();
             _connFactory.RequestedHeartbeat = TimeSpan.FromSeconds(60);
             _connFactory.AutomaticRecoveryEnabled = false;
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsNotType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
 
             _conn.ConnectionShutdown += (_, ea) =>
@@ -111,7 +112,8 @@ public async Task TestMultithreadFloodPublishingAsync()
             _connFactory.DispatchConsumersAsync = true;
             _connFactory.AutomaticRecoveryEnabled = false;
 
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsNotType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
 
             string message = "Hello from test TestMultithreadFloodPublishing";
diff --git a/projects/Test/AsyncIntegration/TestMessageCountAsync.cs b/projects/Test/AsyncIntegration/TestMessageCountAsync.cs
@@ -50,7 +50,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestPassiveDeclareAsync.cs b/projects/Test/AsyncIntegration/TestPassiveDeclareAsync.cs
@@ -52,7 +52,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestPublishSharedChannelAsync.cs b/projects/Test/AsyncIntegration/TestPublishSharedChannelAsync.cs
@@ -67,8 +67,9 @@ public async Task MultiThreadPublishOnSharedChannel()
             var cf = CreateConnectionFactory();
             cf.AutomaticRecoveryEnabled = false;
 
-            using (IConnection conn = cf.CreateConnection())
+            using (IConnection conn = await cf.CreateConnectionAsync())
             {
+                Assert.IsNotType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(conn);
                 conn.ConnectionShutdown += HandleConnectionShutdown;
 
                 using (IChannel channel = await conn.CreateChannelAsync())
diff --git a/projects/Test/AsyncIntegration/TestPublisherConfirmsAsync.cs b/projects/Test/AsyncIntegration/TestPublisherConfirmsAsync.cs
@@ -59,7 +59,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 
diff --git a/projects/Test/AsyncIntegration/TestQueueDeclareAsync.cs b/projects/Test/AsyncIntegration/TestQueueDeclareAsync.cs
@@ -53,7 +53,8 @@ protected override void SetUp()
         public async Task InitializeAsync()
         {
             _connFactory = CreateConnectionFactory();
-            _conn = _connFactory.CreateConnection();
+            _conn = await _connFactory.CreateConnectionAsync();
+            Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);
             _channel = await _conn.CreateChannelAsync();
         }
 

Original file line number	Diff line number	Diff line change
`@@ -80,12 +80,14 @@ internal AutorecoveringConnection(ConnectionConfig config, IEndpointResolver end`
`80`	`80`
`81`	`81`	`internal IConnection Open()`
`82`	`82`	`{`
`83`		`- return InnerConnection.Open();`
	`83`	`+ InnerConnection.Open();`
	`84`	`+ return this;`
`84`	`85`	`}`
`85`	`86`
`86`		`- internal ValueTask<IConnection> OpenAsync()`
	`87`	`+ internal async ValueTask<IConnection> OpenAsync()`
`87`	`88`	`{`
`88`		`- return InnerConnection.OpenAsync();`
	`89`	`+ await InnerConnection.OpenAsync();`
	`90`	`+ return this;`
`89`	`91`	`}`
`90`	`92`
`91`	`93`	`public event EventHandler<EventArgs> RecoverySucceeded`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,8 @@ public async Task InitializeAsync()`
`56`	`56`	`{`
`57`	`57`	`_connFactory = CreateConnectionFactory();`
`58`	`58`	`_connFactory.DispatchConsumersAsync = true;`
`59`		`- _conn = _connFactory.CreateConnection();`
	`59`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`60`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`60`	`61`	`_channel = await _conn.CreateChannelAsync();`
`61`	`62`	`}`
`62`	`63`
`@@ -118,22 +119,23 @@ await TestExceptionHandlingWith(consumer, (ch, q, c, ct) =>`
`118`	`119`	`protected async Task TestExceptionHandlingWith(IBasicConsumer consumer,`
`119`	`120`	`Func<IChannel, string, IBasicConsumer, string, ValueTask> action)`
`120`	`121`	`{`
`121`		`- var semaphore = new SemaphoreSlim(0, 1);`
`122`		`- bool notified = false;`
	`122`	`+ var waitSpan = TimeSpan.FromSeconds(2);`
	`123`	`+ var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);`
	`124`	`+ var cts = new CancellationTokenSource(waitSpan);`
	`125`	`+ cts.Token.Register(() => tcs.TrySetResult(false));`
	`126`	`+`
`123`	`127`	`string q = await _channel.QueueDeclareAsync(string.Empty, false, false, true, false, null);`
`124`	`128`	`_channel.CallbackException += (ch, evt) =>`
`125`	`129`	`{`
`126`	`130`	`if (evt.Exception == TestException)`
`127`	`131`	`{`
`128`		`- notified = true;`
`129`		`- semaphore.Release();`
	`132`	`+ tcs.SetResult(true);`
`130`	`133`	`}`
`131`	`134`	`};`
`132`	`135`
`133`	`136`	`string tag = await _channel.BasicConsumeAsync(q, true, string.Empty, false, false, null, consumer);`
`134`	`137`	`await action(_channel, q, consumer, tag);`
`135`		`- Assert.True(await semaphore.WaitAsync(TimeSpan.FromSeconds(2)));`
`136`		`- Assert.True(notified);`
	`138`	`+ Assert.True(await tcs.Task);`
`137`	`139`	`}`
`138`	`140`
`139`	`141`	`private class ConsumerFailingOnDelivery : AsyncEventingBasicConsumer`
Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,8 @@ protected override void SetUp()`
`50`	`50`	`public async Task InitializeAsync()`
`51`	`51`	`{`
`52`	`52`	`_connFactory = CreateConnectionFactory();`
`53`		`- _conn = _connFactory.CreateConnection();`
	`53`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`54`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`54`	`55`	`_channel = await _conn.CreateChannelAsync();`
`55`	`56`	`}`
`56`	`57`
Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,8 @@ protected override void SetUp()`
`51`	`51`	`public async Task InitializeAsync()`
`52`	`52`	`{`
`53`	`53`	`_connFactory = CreateConnectionFactory();`
`54`		`- _conn = _connFactory.CreateConnection();`
	`54`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`55`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`55`	`56`	`_channel = await _conn.CreateChannelAsync();`
`56`	`57`	`}`
`57`	`58`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@`
`40`	`40`
`41`	`41`	`namespace Test.AsyncIntegration`
`42`	`42`	`{`
`43`		`- public class TestConcurrentAccessWithSharedConnectionAsync : IntegrationFixture`
	`43`	`+ public class TestConcurrentAccessWithSharedConnectionAsync : IntegrationFixture, IAsyncLifetime`
`44`	`44`	`{`
`45`	`45`	`private const ushort _messageCount = 200;`
`46`	`46`
`@@ -49,14 +49,24 @@ public TestConcurrentAccessWithSharedConnectionAsync(ITestOutputHelper output) :`
`49`	`49`	`}`
`50`	`50`
`51`	`51`	`protected override void SetUp()`
	`52`	`+ {`
	`53`	`+ }`
	`54`	`+`
	`55`	`+ public async Task InitializeAsync()`
`52`	`56`	`{`
`53`	`57`	`_connFactory = CreateConnectionFactory();`
`54`		`- _conn = _connFactory.CreateConnection();`
	`58`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`59`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`55`	`60`	`_conn.ConnectionShutdown += HandleConnectionShutdown;`
`56`	`61`	`// NB: not creating _channel because this test suite doesn't use it.`
`57`	`62`	`Assert.Null(_channel);`
`58`	`63`	`}`
`59`	`64`
	`65`	`+ public Task DisposeAsync()`
	`66`	`+ {`
	`67`	`+ return Task.CompletedTask;`
	`68`	`+ }`
	`69`	`+`
`60`	`70`	`[Fact]`
`61`	`71`	`public Task TestConcurrentChannelOpenAndPublishingWithBlankMessagesAsync()`
`62`	`72`	`{`
Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,8 @@ protected override void SetUp()`
`55`	`55`	`public async Task InitializeAsync()`
`56`	`56`	`{`
`57`	`57`	`_connFactory = CreateConnectionFactory();`
`58`		`- _conn = _connFactory.CreateConnection();`
	`58`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`59`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`59`	`60`	`_channel = await _conn.CreateChannelAsync();`
`60`	`61`	`}`
`61`	`62`
Original file line number	Diff line number	Diff line change
`@@ -52,7 +52,8 @@ protected override void SetUp()`
`52`	`52`	`public async Task InitializeAsync()`
`53`	`53`	`{`
`54`	`54`	`_connFactory = CreateConnectionFactory();`
`55`		`- _conn = _connFactory.CreateConnection();`
	`55`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`56`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`56`	`57`	`_channel = await _conn.CreateChannelAsync();`
`57`	`58`	`}`
`58`	`59`
Original file line number	Diff line number	Diff line change
`@@ -67,8 +67,9 @@ public async Task MultiThreadPublishOnSharedChannel()`
`67`	`67`	`var cf = CreateConnectionFactory();`
`68`	`68`	`cf.AutomaticRecoveryEnabled = false;`
`69`	`69`
`70`		`- using (IConnection conn = cf.CreateConnection())`
	`70`	`+ using (IConnection conn = await cf.CreateConnectionAsync())`
`71`	`71`	`{`
	`72`	`+ Assert.IsNotType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(conn);`
`72`	`73`	`conn.ConnectionShutdown += HandleConnectionShutdown;`
`73`	`74`
`74`	`75`	`using (IChannel channel = await conn.CreateChannelAsync())`
Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,8 @@ protected override void SetUp()`
`59`	`59`	`public async Task InitializeAsync()`
`60`	`60`	`{`
`61`	`61`	`_connFactory = CreateConnectionFactory();`
`62`		`- _conn = _connFactory.CreateConnection();`
	`62`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`63`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`63`	`64`	`_channel = await _conn.CreateChannelAsync();`
`64`	`65`	`}`
`65`	`66`
Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,8 @@ protected override void SetUp()`
`53`	`53`	`public async Task InitializeAsync()`
`54`	`54`	`{`
`55`	`55`	`_connFactory = CreateConnectionFactory();`
`56`		`- _conn = _connFactory.CreateConnection();`
	`56`	`+ _conn = await _connFactory.CreateConnectionAsync();`
	`57`	`+ Assert.IsType<RabbitMQ.Client.Framing.Impl.AutorecoveringConnection>(_conn);`
`57`	`58`	`_channel = await _conn.CreateChannelAsync();`
`58`	`59`	`}`
`59`	`60`