fixup

Author: lukebakken

projects/Test/Integration/TestAsyncConsumer.cs

@@ -53,6 +53,7 @@ public TestAsyncConsumer(ITestOutputHelper output)
         public async Task TestBasicRoundtripConcurrent()
         {
             QueueDeclareOk q = await _channel.QueueDeclareAsync();
+
             string publish1 = GetUniqueString(1024);
             byte[] body = _encoding.GetBytes(publish1);
             await _channel.BasicPublishAsync("", q.QueueName, body);
@@ -129,8 +130,8 @@ public async Task TestBasicRoundtripConcurrent()
             }
             finally
             {
-                tokenSource.Dispose();
                 ctsr.Dispose();
+                tokenSource.Dispose();
             }
         }
 
@@ -188,8 +189,28 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                         {
                             using (IConnection publishConn = await _connFactory.CreateConnectionAsync())
                             {
+                                publishConn.ConnectionShutdown += (o, ea) =>
+                                {
+                                    HandleConnectionShutdown(publishConn, ea, (args) =>
+                                    {
+                                        if (args.Initiator == ShutdownInitiator.Peer)
+                                        {
+                                            MaybeSetException(ea, publish1SyncSource, publish2SyncSource);
+                                        }
+                                    });
+                                };
                                 using (IChannel publishChannel = await publishConn.CreateChannelAsync())
                                 {
+                                    publishChannel.ChannelShutdown += (o, ea) =>
+                                    {
+                                        HandleChannelShutdown(publishChannel, ea, (args) =>
+                                        {
+                                            if (args.Initiator == ShutdownInitiator.Peer)
+                                            {
+                                                MaybeSetException(ea, publish1SyncSource, publish2SyncSource);
+                                            }
+                                        });
+                                    };
                                     await publishChannel.ConfirmSelectAsync();
 
                                     for (int i = 0; i < publish_total; i++)
@@ -210,8 +231,29 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                         {
                             using (IConnection consumeConn = await _connFactory.CreateConnectionAsync())
                             {
+                                consumeConn.ConnectionShutdown += (o, ea) =>
+                                {
+                                    HandleConnectionShutdown(consumeConn, ea, (args) =>
+                                    {
+                                        if (args.Initiator == ShutdownInitiator.Peer)
+                                        {
+                                            MaybeSetException(ea, publish1SyncSource, publish2SyncSource);
+                                        }
+                                    });
+                                };
                                 using (IChannel consumeChannel = await consumeConn.CreateChannelAsync())
                                 {
+                                    consumeChannel.ChannelShutdown += (o, ea) =>
+                                    {
+                                        HandleChannelShutdown(consumeChannel, ea, (args) =>
+                                        {
+                                            if (args.Initiator == ShutdownInitiator.Peer)
+                                            {
+                                                MaybeSetException(ea, publish1SyncSource, publish2SyncSource);
+                                            }
+                                        });
+                                    };
+
                                     var consumer = new AsyncEventingBasicConsumer(consumeChannel);
 
                                     int publish1_count = 0;
@@ -254,7 +296,6 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
 
                 await AssertRanToCompletion(publishTask);
 
-                // ensure we get a delivery
                 await AssertRanToCompletion(publish1SyncSource.Task, publish2SyncSource.Task);
                 consumerSyncSource.TrySetResult(true);
 
@@ -266,8 +307,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
             }
             finally
             {
-                tokenSource.Dispose();
                 ctsr.Dispose();
+                tokenSource.Dispose();
             }
         }
 
@@ -367,8 +408,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
             finally
             {
                 await _channel.QueueDeleteAsync(queue: queueName);
-                cancellationTokenSource.Dispose();
                 ctsr.Dispose();
+                cancellationTokenSource.Dispose();
             }
         }