Test refactor

Author: lukebakken

projects/Unit/TestConnectionRecovery.cs

@@ -48,19 +48,21 @@ namespace RabbitMQ.Client.Unit
     public class TestConnectionRecovery : IntegrationFixture
     {
         private readonly byte[] _messageBody;
-        private readonly ushort _totalMessageCount = 64;
-        private string _queueName = "luke-test-queue";
+        private readonly ushort _totalMessageCount = 1024;
+        private readonly ushort _closeAtCount = 16;
+        private string _queueName;
 
         public TestConnectionRecovery()
         {
             var rnd = new Random();
-            _messageBody = new byte[1024];
+            _messageBody = new byte[4096];
             rnd.NextBytes(_messageBody);
         }
 
         [SetUp]
         public override void Init()
         {
+            _queueName = $"TestConnectionRecovery-queue-{Guid.NewGuid()}";
             Conn = CreateAutorecoveringConnection();
             Model = Conn.CreateModel();
             Model.QueueDelete(_queueName);
@@ -73,10 +75,10 @@ public void CleanUp()
         }
 
         [Test]
-        public void LukeTest()
+        public void TestBasicAckAfterChannelRecovery()
         {
             var allMessagesSeenLatch = new ManualResetEventSlim(false);
-            var cons = new LukeAckingBasicConsumer(Model, _totalMessageCount, allMessagesSeenLatch);
+            var cons = new AckingBasicConsumer(Model, _totalMessageCount, allMessagesSeenLatch);
 
             string queueName = Model.QueueDeclare(_queueName, false, false, false, null).QueueName;
             Assert.AreEqual(queueName, _queueName);
@@ -87,42 +89,56 @@ public void LukeTest()
             ManualResetEventSlim sl = PrepareForShutdown(Conn);
             ManualResetEventSlim rl = PrepareForRecovery(Conn);
 
-            using (IAutorecoveringConnection publishingConn = CreateAutorecoveringConnection())
-            {
-                using (IModel publishingModel = publishingConn.CreateModel())
-                {
-                    for (ushort i = 0; i < _totalMessageCount; i++)
-                    {
-                        if (i == 16)
-                        {
-                            CloseConnection(Conn);
-                        }
-                        publishingModel.BasicPublish(string.Empty, queueName, null, _messageBody);
-                    }
-                }
-            }
+            PublishMessagesWhileClosingConn(queueName);
 
-            // System.Console.Error.WriteLine("@@@@@@@@ waiting for shutdown...");
             Wait(sl);
-            // System.Console.Error.WriteLine("@@@@@@@@ waiting for shutdown...DONE");
-            // System.Console.Error.WriteLine("@@@@@@@@ waiting for recovery...");
             Wait(rl);
-            // System.Console.Error.WriteLine("@@@@@@@@ waiting for recovery...DONE");
-            // System.Console.Error.WriteLine("@@@@@@@@ waiting for the rest of the messages...");
             Wait(allMessagesSeenLatch);
-            // System.Console.Error.WriteLine("@@@@@@@@ waiting for the rest of the messages...DONE");
         }
 
-        /*
         [Test]
-        public void TestBasicAckAfterChannelRecovery()
+        public void TestBasicNackAfterChannelRecovery()
         {
-            var latch = new ManualResetEventSlim(false);
-            var cons = new AckingBasicConsumer(Model, latch, CloseAndWaitForRecovery);
+            var allMessagesSeenLatch = new ManualResetEventSlim(false);
+            var cons = new NackingBasicConsumer(Model, _totalMessageCount, allMessagesSeenLatch);
+
+            string queueName = Model.QueueDeclare(_queueName, false, false, false, null).QueueName;
+            Assert.AreEqual(queueName, _queueName);
+
+            Model.BasicQos(0, 1, false);
+            string consumerTag = Model.BasicConsume(queueName, false, cons);
 
-            TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
+            ManualResetEventSlim sl = PrepareForShutdown(Conn);
+            ManualResetEventSlim rl = PrepareForRecovery(Conn);
+
+            PublishMessagesWhileClosingConn(queueName);
+
+            Wait(sl);
+            Wait(rl);
+            Wait(allMessagesSeenLatch);
+        }
+
+        [Test]
+        public void TestBasicRejectAfterChannelRecovery()
+        {
+            var allMessagesSeenLatch = new ManualResetEventSlim(false);
+            var cons = new RejectingBasicConsumer(Model, _totalMessageCount, allMessagesSeenLatch);
+
+            string queueName = Model.QueueDeclare(_queueName, false, false, false, null).QueueName;
+            Assert.AreEqual(queueName, _queueName);
+
+            Model.BasicQos(0, 1, false);
+            string consumerTag = Model.BasicConsume(queueName, false, cons);
+
+            ManualResetEventSlim sl = PrepareForShutdown(Conn);
+            ManualResetEventSlim rl = PrepareForRecovery(Conn);
+
+            PublishMessagesWhileClosingConn(queueName);
+
+            Wait(sl);
+            Wait(rl);
+            Wait(allMessagesSeenLatch);
         }
-        */
 
         [Test]
         public void TestBasicAckAfterBasicGetAndChannelRecovery()
@@ -131,7 +147,7 @@ public void TestBasicAckAfterBasicGetAndChannelRecovery()
             Model.QueueDeclare(q, false, false, false, null);
             // create an offset
             IBasicProperties bp = Model.CreateBasicProperties();
-            Model.BasicPublish("", q, bp, new byte[] { });
+            Model.BasicPublish("", q, bp, _messageBody);
             Thread.Sleep(50);
             BasicGetResult g = Model.BasicGet(q, false);
             CloseAndWaitForRecovery();
@@ -155,8 +171,7 @@ public void TestBasicAckEventHandlerRecovery()
             CloseAndWaitForRecovery();
             Assert.IsTrue(Model.IsOpen);
 
-            byte[] mb = RandomMessageBody();
-            WithTemporaryNonExclusiveQueue(Model, (m, q) => m.BasicPublish("", q, null, mb));
+            WithTemporaryNonExclusiveQueue(Model, (m, q) => m.BasicPublish("", q, null, _messageBody));
             Wait(latch);
         }
 
@@ -265,26 +280,6 @@ public void TestBasicModelRecoveryOnServerRestart()
             Assert.IsTrue(Model.IsOpen);
         }
 
-        /*
-        [Test]
-        public void TestBasicNackAfterChannelRecovery()
-        {
-            var latch = new ManualResetEventSlim(false);
-            var cons = new NackingBasicConsumer(Model, latch, CloseAndWaitForRecovery);
-
-            TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
-        }
-
-        [Test]
-        public void TestBasicRejectAfterChannelRecovery()
-        {
-            var latch = new ManualResetEventSlim(false);
-            var cons = new RejectingBasicConsumer(Model, latch, CloseAndWaitForRecovery);
-
-            TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
-        }
-        */
-
         [Test]
         public void TestBlockedListenersRecovery()
         {
@@ -823,7 +818,7 @@ public void TestRecoverTopologyOnDisposedChannel()
             var latch = new ManualResetEventSlim(false);
             cons.Received += (s, args) => latch.Set();
 
-            Model.BasicPublish("", q, null, ReadOnlyMemory<byte>.Empty);
+            Model.BasicPublish("", q, null, _messageBody);
             Wait(latch);
 
             Model.QueueUnbind(q, x, rk);
@@ -861,7 +856,7 @@ public void TestPublishRpcRightAfterReconnect()
             {
                 try
                 {
-                    Model.BasicPublish(string.Empty, testQueueName, false, properties, ReadOnlyMemory<byte>.Empty);
+                    Model.BasicPublish(string.Empty, testQueueName, false, properties, _messageBody);
                 }
                 catch (Exception e)
                 {
@@ -1014,8 +1009,7 @@ internal void AssertExchangeRecovery(IModel m, string x)
             {
                 string rk = "routing-key";
                 m.QueueBind(q, x, rk);
-                byte[] mb = RandomMessageBody();
-                m.BasicPublish(x, rk, null, mb);
+                m.BasicPublish(x, rk, null, _messageBody);
 
                 Assert.IsTrue(WaitForConfirms(m));
                 m.ExchangeDeclarePassive(x);
@@ -1033,8 +1027,7 @@ internal void AssertQueueRecovery(IModel m, string q, bool exclusive)
             m.QueueDeclarePassive(q);
             QueueDeclareOk ok1 = m.QueueDeclare(q, false, exclusive, false, null);
             Assert.AreEqual(ok1.MessageCount, 0);
-            byte[] mb = RandomMessageBody();
-            m.BasicPublish("", q, null, mb);
+            m.BasicPublish("", q, null, _messageBody);
             Assert.IsTrue(WaitForConfirms(m));
             QueueDeclareOk ok2 = m.QueueDeclare(q, false, exclusive, false, null);
             Assert.AreEqual(ok2.MessageCount, 1);
@@ -1122,30 +1115,6 @@ internal void RestartServerAndWaitForRecovery(IAutorecoveringConnection conn)
             Wait(rl);
         }
 
-        /*
-        internal void TestDelayedBasicAckNackAfterChannelRecovery(TestBasicConsumer1 cons, ManualResetEventSlim latch)
-        {
-            string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
-            int n = 30;
-            Model.BasicQos(0, 1, false);
-            Model.BasicConsume(q, false, cons);
-
-            AutorecoveringConnection publishingConn = CreateAutorecoveringConnection();
-            IModel publishingModel = publishingConn.CreateModel();
-
-            byte[] mb = RandomMessageBody();
-            for (int i = 0; i < n; i++)
-            {
-                publishingModel.BasicPublish("", q, null, mb);
-            }
-
-            Wait(latch, TimeSpan.FromSeconds(20));
-            Model.QueueDelete(q);
-            publishingModel.Close();
-            publishingConn.Close();
-        }
-        */
-
         internal void WaitForRecovery()
         {
             Wait(PrepareForRecovery((AutorecoveringConnection)Conn));
@@ -1166,46 +1135,28 @@ internal void WaitForShutdown(IConnection conn)
             Wait(PrepareForShutdown(conn));
         }
 
-        public class LukeAckingBasicConsumer : DefaultBasicConsumer
+        internal void PublishMessagesWhileClosingConn(string queueName)
         {
-            private readonly ManualResetEventSlim _allMessagesSeenLatch;
-            private readonly ushort _totalMessageCount;
-            private ushort _counter = 0;
-
-            public LukeAckingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
-                : base(model)
-            {
-                _totalMessageCount = totalMessageCount;
-                _allMessagesSeenLatch = allMessagesSeenLatch;
-            }
-
-            public override void HandleBasicDeliver(string consumerTag,
-                ulong deliveryTag,
-                bool redelivered,
-                string exchange,
-                string routingKey,
-                IBasicProperties properties,
-                ReadOnlyMemory<byte> body)
+            using (IAutorecoveringConnection publishingConn = CreateAutorecoveringConnection())
             {
-                try
-                {
-                    Model.BasicAck(deliveryTag, false);
-                }
-                finally
+                using (IModel publishingModel = publishingConn.CreateModel())
                 {
-                    ++_counter;
-                    if (_counter >= _totalMessageCount)
+                    for (ushort i = 0; i < _totalMessageCount; i++)
                     {
-                        _allMessagesSeenLatch.Set();
+                        if (i == _closeAtCount)
+                        {
+                            CloseConnection(Conn);
+                        }
+                        publishingModel.BasicPublish(string.Empty, queueName, null, _messageBody);
                     }
                 }
             }
         }
 
-        public class AckingBasicConsumer : TestBasicConsumer1
+        public class AckingBasicConsumer : TestBasicConsumer
         {
-            public AckingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)
-                : base(model, latch, fn)
+            public AckingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
+                : base(model, totalMessageCount, allMessagesSeenLatch)
             {
             }
 
@@ -1215,10 +1166,10 @@ public override void PostHandleDelivery(ulong deliveryTag)
             }
         }
 
-        public class NackingBasicConsumer : TestBasicConsumer1
+        public class NackingBasicConsumer : TestBasicConsumer
         {
-            public NackingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)
-                : base(model, latch, fn)
+            public NackingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
+                : base(model, totalMessageCount, allMessagesSeenLatch)
             {
             }
 
@@ -1228,10 +1179,10 @@ public override void PostHandleDelivery(ulong deliveryTag)
             }
         }
 
-        public class RejectingBasicConsumer : TestBasicConsumer1
+        public class RejectingBasicConsumer : TestBasicConsumer
         {
-            public RejectingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)
-                : base(model, latch, fn)
+            public RejectingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
+                : base(model, totalMessageCount, allMessagesSeenLatch)
             {
             }
 
@@ -1241,16 +1192,17 @@ public override void PostHandleDelivery(ulong deliveryTag)
             }
         }
 
-        public class TestBasicConsumer1 : DefaultBasicConsumer
+        public class TestBasicConsumer : DefaultBasicConsumer
         {
-            private readonly Action _action;
-            private readonly ManualResetEventSlim _latch;
+            private readonly ManualResetEventSlim _allMessagesSeenLatch;
+            private readonly ushort _totalMessageCount;
+            private ushort _counter = 0;
 
-            public TestBasicConsumer1(IModel model, ManualResetEventSlim latch, Action fn)
+            public TestBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
                 : base(model)
             {
-                _latch = latch;
-                _action = fn;
+                _totalMessageCount = totalMessageCount;
+                _allMessagesSeenLatch = allMessagesSeenLatch;
             }
 
             public override void HandleBasicDeliver(string consumerTag,
@@ -1263,23 +1215,15 @@ public override void HandleBasicDeliver(string consumerTag,
             {
                 try
                 {
-                    if (deliveryTag == 5)
-                    {
-                        _action();
-                    }
-
-                    /*
-                     * Note: wait for one more message to be delivered before
-                     * setting this reset event
-                     */
-                    if (deliveryTag == 6)
-                    {
-                        _latch.Set();
-                    }
+                    PostHandleDelivery(deliveryTag);
                 }
                 finally
                 {
-                    PostHandleDelivery(deliveryTag);
+                    ++_counter;
+                    if (_counter >= _totalMessageCount)
+                    {
+                        _allMessagesSeenLatch.Set();
+                    }
                 }
             }