* Use separate publish and consume connections.

Author: lukebakken

projects/Test/Integration/TestFloodPublishing.cs

@@ -163,29 +163,34 @@ public async Task TestMultithreadFloodPublishing()
             Task pub = Task.Run(async () =>
             {
                 bool stop = false;
-                using (IChannel pubCh = await _conn.CreateChannelAsync())
+                using (IConnection publishConnection = await _connFactory.CreateConnectionAsync())
                 {
-                    await pubCh.ConfirmSelectAsync();
-
-                    pubCh.ChannelShutdown += (o, ea) =>
+                    using (IChannel publishChannel = await publishConnection.CreateChannelAsync())
                     {
-                        HandleChannelShutdown(pubCh, ea, (args) =>
+                        await publishChannel.ConfirmSelectAsync();
+
+                        publishChannel.ChannelShutdown += (o, ea) =>
                         {
-                            if (args.Initiator == ShutdownInitiator.Peer)
+                            HandleChannelShutdown(publishChannel, ea, (args) =>
                             {
-                                stop = true;
-                                tcs.TrySetResult(false);
-                            }
-                        });
-                    };
+                                if (args.Initiator == ShutdownInitiator.Peer)
+                                {
+                                    stop = true;
+                                    tcs.TrySetResult(false);
+                                }
+                            });
+                        };
+
+                        for (int i = 0; i < publishCount && false == stop; i++)
+                        {
+                            await publishChannel.BasicPublishAsync(string.Empty, queueName, sendBody, true);
+                        }
 
-                    for (int i = 0; i < publishCount && false == stop; i++)
-                    {
-                        await pubCh.BasicPublishAsync(string.Empty, queueName, sendBody, true);
+                        await publishChannel.WaitForConfirmsOrDieAsync();
+                        await publishChannel.CloseAsync();
                     }
 
-                    await pubCh.WaitForConfirmsOrDieAsync();
-                    await pubCh.CloseAsync();
+                    await publishConnection.CloseAsync();
                 }
             });
 
@@ -197,37 +202,42 @@ public async Task TestMultithreadFloodPublishing()
 
             try
             {
-                using (IChannel consumeCh = await _conn.CreateChannelAsync())
+                using (IConnection consumeConnection = await _connFactory.CreateConnectionAsync())
                 {
-                    consumeCh.ChannelShutdown += (o, ea) =>
+                    using (IChannel consumeChannel = await consumeConnection.CreateChannelAsync())
                     {
-                        HandleChannelShutdown(consumeCh, ea, (args) =>
+                        consumeChannel.ChannelShutdown += (o, ea) =>
                         {
-                            if (args.Initiator == ShutdownInitiator.Peer)
+                            HandleChannelShutdown(consumeChannel, ea, (args) =>
                             {
-                                tcs.TrySetResult(false);
+                                if (args.Initiator == ShutdownInitiator.Peer)
+                                {
+                                    tcs.TrySetResult(false);
+                                }
+                            });
+                        };
+
+                        var consumer = new AsyncEventingBasicConsumer(consumeChannel);
+                        consumer.Received += async (o, a) =>
+                        {
+                            string receivedMessage = _encoding.GetString(a.Body.ToArray());
+                            Assert.Equal(message, receivedMessage);
+                            if (Interlocked.Increment(ref receivedCount) == publishCount)
+                            {
+                                tcs.SetResult(true);
                             }
-                        });
-                    };
+                            await Task.Yield();
+                        };
 
-                    var consumer = new AsyncEventingBasicConsumer(consumeCh);
-                    consumer.Received += async (o, a) =>
-                    {
-                        string receivedMessage = _encoding.GetString(a.Body.ToArray());
-                        Assert.Equal(message, receivedMessage);
-                        if (Interlocked.Increment(ref receivedCount) == publishCount)
-                        {
-                            tcs.SetResult(true);
-                        }
-                        await Task.Yield();
-                    };
+                        await consumeChannel.BasicConsumeAsync(queue: queueName, autoAck: true,
+                            consumerTag: string.Empty, noLocal: false, exclusive: false,
+                            arguments: null, consumer: consumer);
 
-                    await consumeCh.BasicConsumeAsync(queue: queueName, autoAck: true,
-                        consumerTag: string.Empty, noLocal: false, exclusive: false,
-                        arguments: null, consumer: consumer);
+                        Assert.True(await tcs.Task);
+                        await consumeChannel.CloseAsync();
+                    }
 
-                    Assert.True(await tcs.Task);
-                    await consumeCh.CloseAsync();
+                    await consumeConnection.CloseAsync();
                 }
 
                 await pub;