Concurrent dispatch as opt-in

Author: danielmarbach

projects/Unit/TestAsyncConsumer.cs

@@ -39,6 +39,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -82,36 +83,55 @@ public void TestBasicRoundtrip()
         }
 
         [Test]
-        public void TestBasicRoundtripConcurrent()
+        public async Task TestBasicRoundtripConcurrent()
         {
             var cf = new ConnectionFactory{ DispatchConsumersAsync = true, ProcessingConcurrency = 2 };
             using(IConnection c = cf.CreateConnection())
             using(IModel m = c.CreateModel())
             {
                 QueueDeclareOk q = m.QueueDeclare();
                 IBasicProperties bp = m.CreateBasicProperties();
-                var body = System.Text.Encoding.UTF8.GetBytes("async-hi-1");
+                const string publish1 = "async-hi-1";
+                var body = Encoding.UTF8.GetBytes(publish1);
                 m.BasicPublish("", q.QueueName, bp, body);
-                body = System.Text.Encoding.UTF8.GetBytes("async-hi-2");
+                const string publish2 = "async-hi-2";
+                body = Encoding.UTF8.GetBytes(publish2);
                 m.BasicPublish("", q.QueueName, bp, body);
+
                 var consumer = new AsyncEventingBasicConsumer(m);
-                var are = new CountdownEvent(2);
+
+                var publish1SyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                var publish2SyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                var maximumWaitTime = TimeSpan.FromSeconds(5);
+                var tokenSource = new CancellationTokenSource(maximumWaitTime);
+                tokenSource.Token.Register(() =>
+                {
+                    publish1SyncSource.TrySetResult(false);
+                    publish2SyncSource.TrySetResult(false);
+                });
+
                 consumer.Received += async (o, a) =>
                 {
-                    await Task.Delay(500);
-                    are.Signal();
+                    switch (Encoding.UTF8.GetString(a.Body.ToArray()))
+                    {
+                        case publish1:
+                            publish1SyncSource.TrySetResult(true);
+                            await publish2SyncSource.Task;
+                            break;
+                        case publish2:
+                            publish2SyncSource.TrySetResult(true);
+                            await publish1SyncSource.Task;
+                            break;
+                    }
                 };
-                string tag = m.BasicConsume(q.QueueName, true, consumer);
+
+                m.BasicConsume(q.QueueName, true, consumer);
                 // ensure we get a delivery
-                bool waitRes = are.Wait(800);
-                Assert.IsTrue(waitRes);
 
-                are.Reset(1);
-                // unsubscribe and ensure no further deliveries
-                m.BasicCancel(tag);
-                m.BasicPublish("", q.QueueName, bp, body);
-                bool waitResFalse = are.Wait(2000);
-                Assert.IsFalse(waitResFalse);
+                await Task.WhenAll(publish1SyncSource.Task, publish2SyncSource.Task);
+
+                Assert.IsTrue(publish1SyncSource.Task.Result, $"Non concurrent dispatch lead to deadlock after {maximumWaitTime}");
+                Assert.IsTrue(publish2SyncSource.Task.Result, $"Non concurrent dispatch lead to deadlock after {maximumWaitTime}");
             }
         }
 

projects/Unit/TestConsumer.cs

@@ -1,4 +1,7 @@
+using System;
+using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
 using NUnit.Framework;
 using RabbitMQ.Client.Events;
 
@@ -8,36 +11,54 @@ namespace RabbitMQ.Client.Unit
     public class TestConsumer
     {
         [Test]
-        public void TestBasicRoundtripConcurrent()
+        public async Task TestBasicRoundtripConcurrent()
         {
-            var cf = new ConnectionFactory{ ProcessingConcurrency = 2 };
+            var cf = new ConnectionFactory{ ProcessingConcurrency = 1 };
             using(IConnection c = cf.CreateConnection())
             using(IModel m = c.CreateModel())
             {
                 QueueDeclareOk q = m.QueueDeclare();
                 IBasicProperties bp = m.CreateBasicProperties();
-                var body = System.Text.Encoding.UTF8.GetBytes("async-hi-1");
+                const string publish1 = "sync-hi-1";
+                var body = Encoding.UTF8.GetBytes(publish1);
                 m.BasicPublish("", q.QueueName, bp, body);
-                body = System.Text.Encoding.UTF8.GetBytes("async-hi-2");
+                const string publish2 = "sync-hi-2";
+                body = Encoding.UTF8.GetBytes(publish2);
                 m.BasicPublish("", q.QueueName, bp, body);
+
                 var consumer = new EventingBasicConsumer(m);
-                var are = new CountdownEvent(2);
+
+                var publish1SyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                var publish2SyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                var maximumWaitTime = TimeSpan.FromSeconds(5);
+                var tokenSource = new CancellationTokenSource(maximumWaitTime);
+                tokenSource.Token.Register(() =>
+                {
+                    publish1SyncSource.TrySetResult(false);
+                    publish2SyncSource.TrySetResult(false);
+                });
+
                 consumer.Received += (o, a) =>
                 {
-                    Thread.Sleep(500);
-                    are.Signal();
+                    switch (Encoding.UTF8.GetString(a.Body.ToArray()))
+                    {
+                        case publish1:
+                            publish1SyncSource.TrySetResult(true);
+                            publish2SyncSource.Task.GetAwaiter().GetResult();
+                            break;
+                        case publish2:
+                            publish2SyncSource.TrySetResult(true);
+                            publish1SyncSource.Task.GetAwaiter().GetResult();
+                            break;
+                    }
                 };
-                string tag = m.BasicConsume(q.QueueName, true, consumer);
-                // ensure we get a delivery
-                bool waitRes = are.Wait(800);
-                Assert.IsTrue(waitRes);
 
-                are.Reset(1);
-                // unsubscribe and ensure no further deliveries
-                m.BasicCancel(tag);
-                m.BasicPublish("", q.QueueName, bp, body);
-                bool waitResFalse = are.Wait(2000);
-                Assert.IsFalse(waitResFalse);
+                m.BasicConsume(q.QueueName, true, consumer);
+
+                await Task.WhenAll(publish1SyncSource.Task, publish2SyncSource.Task);
+
+                Assert.IsTrue(publish1SyncSource.Task.Result, $"Non concurrent dispatch lead to deadlock after {maximumWaitTime}");
+                Assert.IsTrue(publish2SyncSource.Task.Result, $"Non concurrent dispatch lead to deadlock after {maximumWaitTime}");
             }
         }
     }