* Add concurrency to try to reproduce issue.

Author: lukebakken

projects/Test/Integration/GH/TestGitHubIssues.cs

@@ -333,6 +333,8 @@ await ch.BasicPublishAsync(exchange: string.Empty, routingKey: queueName,
         [Fact]
         public async Task MaybeSomethingUpWithRateLimiter_GH1793()
         {
+            const int messageCount = 16;
+
             _connFactory = new ConnectionFactory
             {
                 AutomaticRecoveryEnabled = true
@@ -343,7 +345,7 @@ public async Task MaybeSomethingUpWithRateLimiter_GH1793()
             var channelOpts = new CreateChannelOptions(
                 publisherConfirmationsEnabled: true,
                 publisherConfirmationTrackingEnabled: true,
-                outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(256)
+                outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(messageCount)
             );
 
             _channel = await _conn.CreateChannelAsync(channelOpts);
@@ -352,22 +354,40 @@ public async Task MaybeSomethingUpWithRateLimiter_GH1793()
             {
                 DeliveryMode = DeliveryModes.Persistent
             };
-            int retryCount = 0;
-            const int maxRetries = 3;
-            while (retryCount <= maxRetries)
+
+            async Task PublishMessagesAsync()
             {
-                try
-                {
-                    var bytes = Encoding.UTF8.GetBytes("message");
-                    await _channel.BasicPublishAsync(string.Empty, string.Empty, true, properties, bytes);
-                    break;
-                }
-                catch (Exception ex)
+                for (int i = 0; i < messageCount; i++)
                 {
-                    retryCount++;
-                    _output.WriteLine("{0} exception: {1}", _testDisplayName, ex);
+                    int retryCount = 0;
+                    const int maxRetries = 3;
+                    while (retryCount <= maxRetries)
+                    {
+                        try
+                        {
+                            byte[] bytes = Encoding.UTF8.GetBytes("message");
+                            await _channel.BasicPublishAsync(string.Empty, string.Empty, true, properties, bytes);
+                            break;
+                        }
+                        catch (SemaphoreFullException ex0)
+                        {
+                            _output.WriteLine("{0} ex: {1}", _testDisplayName, ex0);
+                            retryCount++;
+                        }
+                        catch (PublishException)
+                        {
+                            retryCount++;
+                        }
+                    }
                 }
             }
+
+            var publishTasks = new List<Task>();
+            for (int i = 0; i < messageCount; i++)
+            {
+                publishTasks.Add(Task.Run(PublishMessagesAsync));
+            }
+            await Task.WhenAll(publishTasks);
         }
     }
 }