* Add queue arguments to test to simulate nacks.

Author: lukebakken

projects/Test/Integration/TestFloodPublishing.cs

@@ -30,6 +30,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Threading;
@@ -43,7 +44,7 @@ namespace Test.Integration
 {
     public class TestFloodPublishing : IntegrationFixture
     {
-        private static readonly TimeSpan FiveSeconds = TimeSpan.FromSeconds(5);
+        private static readonly TimeSpan ElapsedMax = TimeSpan.FromSeconds(10);
         private readonly byte[] _body = GetRandomBody(2048);
 
         public TestFloodPublishing(ITestOutputHelper output) : base(output)
@@ -92,36 +93,82 @@ public async Task TestUnthrottledFloodPublishing()
                 return Task.CompletedTask;
             };
 
+            var queueArguments = new Dictionary<string, object>
+            {
+                ["x-max-length"] = 131072,
+                ["x-overflow"] = "reject-publish"
+            };
+
+            QueueDeclareOk q = await _channel.QueueDeclareAsync(queue: string.Empty,
+                    passive: false, durable: false, exclusive: true, autoDelete: true, arguments: queueArguments);
+            string queueName = q.QueueName;
+
+            var exceptions = new ConcurrentBag<Exception>();
             var stopwatch = Stopwatch.StartNew();
             int publishCount = 0;
             try
             {
                 var tasks = new List<Task>();
-                for (int j = 0; j < 64; j++)
+                for (int j = 0; j < 8; j++)
                 {
                     tasks.Add(Task.Run(async () =>
                     {
-                        var publishTasks = new List<Task>();
-                        for (int i = 0; i < 65536 * 2; i++)
+                        var publishTasks = new List<ValueTask>();
+                        for (int i = 0; i < 65536; i++)
                         {
-                            if (stopwatch.Elapsed > FiveSeconds)
+                            if (stopwatch.Elapsed > ElapsedMax)
                             {
-                                await Task.WhenAll(publishTasks).WaitAsync(ShortSpan);
+                                foreach (ValueTask pt in publishTasks)
+                                {
+                                    try
+                                    {
+                                        await pt;
+                                    }
+                                    catch (Exception ex)
+                                    {
+                                        exceptions.Add(ex);
+                                    }
+                                }
                                 publishTasks.Clear();
-                                break;
+                                return;
                             }
 
                             Interlocked.Increment(ref publishCount);
-                            publishTasks.Add(_channel.BasicPublishAsync(CachedString.Empty, CachedString.Empty, _body).AsTask());
+                            publishTasks.Add(_channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName, mandatory: true,
+                                body: _body));
 
-                            if (i % 500 == 0)
+                            if (i % 128 == 0)
                             {
-                                await Task.WhenAll(publishTasks).WaitAsync(ShortSpan);
+                                foreach (ValueTask pt in publishTasks)
+                                {
+                                    try
+                                    {
+                                        await pt;
+                                    }
+                                    catch (Exception ex)
+                                    {
+                                        exceptions.Add(ex);
+                                    }
+                                }
                                 publishTasks.Clear();
                             }
                         }
+
+                        foreach (ValueTask pt in publishTasks)
+                        {
+                            try
+                            {
+                                await pt;
+                            }
+                            catch (Exception ex)
+                            {
+                                exceptions.Add(ex);
+                            }
+                        }
+                        publishTasks.Clear();
                     }));
                 }
+
                 await Task.WhenAll(tasks).WaitAsync(WaitSpan);
             }
             finally
@@ -131,9 +178,11 @@ public async Task TestUnthrottledFloodPublishing()
 
             Assert.True(_conn.IsOpen);
             Assert.False(sawUnexpectedShutdown);
-            if (IsVerbose)
+            // if (IsVerbose)
+            if (true)
             {
-                _output.WriteLine("[INFO] published {0} messages in {1}", publishCount, stopwatch.Elapsed);
+                _output.WriteLine("[INFO] published {0} messages in {1}, exceptions: {2}",
+                    publishCount, stopwatch.Elapsed, exceptions.Count);
             }
         }