Add an ad-hoc test to try and figure out what's up.

Author: lukebakken

projects/RabbitMQ.Client/client/impl/IncomingCommand.cs

@@ -29,5 +29,10 @@ public void Dispose()
                 ArrayPool<byte>.Shared.Return(_rentedArray);
             }
         }
+
+        public override string ToString()
+        {
+            return $"IncomingCommand Method={Method.ProtocolMethodName}, Body.Length={Body.Length}";
+        }
     }
 }

projects/Unit/TestConnectionRecovery.cs

@@ -44,36 +44,78 @@
 
 namespace RabbitMQ.Client.Unit
 {
-    class DisposableConnection : IDisposable
+    [TestFixture]
+    public class TestConnectionRecovery : IntegrationFixture
     {
-        public DisposableConnection(AutorecoveringConnection c)
-        {
-            Connection = c;
-        }
+        private readonly byte[] _messageBody;
+        private readonly ushort _totalMessageCount = 64;
+        private string _queueName = "luke-test-queue";
 
-        public AutorecoveringConnection Connection { get; private set; }
-
-        public void Dispose()
+        public TestConnectionRecovery()
         {
-            Connection.Close();
+            var rnd = new Random();
+            _messageBody = new byte[1024];
+            rnd.NextBytes(_messageBody);
         }
-    }
-    [TestFixture]
-    public class TestConnectionRecovery : IntegrationFixture
-    {
+
         [SetUp]
         public override void Init()
         {
             Conn = CreateAutorecoveringConnection();
             Model = Conn.CreateModel();
+            Model.QueueDelete(_queueName);
         }
 
         [TearDown]
         public void CleanUp()
         {
+            Model.QueueDelete(_queueName);
+            Model.Close();
             Conn.Close();
         }
 
+        [Test]
+        public void LukeTest()
+        {
+            var allMessagesSeenLatch = new ManualResetEventSlim(false);
+            var cons = new LukeAckingBasicConsumer(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);
+
+            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);
+                    }
+                }
+            }
+
+            // 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()
         {
@@ -82,6 +124,7 @@ public void TestBasicAckAfterChannelRecovery()
 
             TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
         }
+        */
 
         [Test]
         public void TestBasicAckAfterBasicGetAndChannelRecovery()
@@ -224,6 +267,7 @@ public void TestBasicModelRecoveryOnServerRestart()
             Assert.IsTrue(Model.IsOpen);
         }
 
+        /*
         [Test]
         public void TestBasicNackAfterChannelRecovery()
         {
@@ -241,6 +285,7 @@ public void TestBasicRejectAfterChannelRecovery()
 
             TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
         }
+        */
 
         [Test]
         public void TestBlockedListenersRecovery()
@@ -1033,25 +1078,29 @@ internal void CloseAndWaitForRecovery(AutorecoveringConnection conn)
             Wait(rl);
         }
 
-        internal void CloseAndWaitForShutdown(AutorecoveringConnection conn)
+        internal void CloseAndWaitForShutdown(IAutorecoveringConnection conn)
         {
             ManualResetEventSlim sl = PrepareForShutdown(conn);
             CloseConnection(conn);
             Wait(sl);
         }
 
-        internal ManualResetEventSlim PrepareForRecovery(AutorecoveringConnection conn)
+        internal ManualResetEventSlim PrepareForRecovery(IConnection conn)
         {
             var latch = new ManualResetEventSlim(false);
-            conn.RecoverySucceeded += (source, ea) => latch.Set();
+
+            IAutorecoveringConnection aconn = conn as IAutorecoveringConnection;
+            aconn.RecoverySucceeded += (source, ea) => latch.Set();
 
             return latch;
         }
 
         internal ManualResetEventSlim PrepareForShutdown(IConnection conn)
         {
             var latch = new ManualResetEventSlim(false);
-            conn.ConnectionShutdown += (c, args) => latch.Set();
+
+            IAutorecoveringConnection aconn = conn as IAutorecoveringConnection;
+            aconn.ConnectionShutdown += (c, args) => latch.Set();
 
             return latch;
         }
@@ -1063,10 +1112,10 @@ protected override void ReleaseResources()
 
         internal void RestartServerAndWaitForRecovery()
         {
-            RestartServerAndWaitForRecovery((AutorecoveringConnection)Conn);
+            RestartServerAndWaitForRecovery((IAutorecoveringConnection)Conn);
         }
 
-        internal void RestartServerAndWaitForRecovery(AutorecoveringConnection conn)
+        internal void RestartServerAndWaitForRecovery(IAutorecoveringConnection conn)
         {
             ManualResetEventSlim sl = PrepareForShutdown(conn);
             ManualResetEventSlim rl = PrepareForRecovery(conn);
@@ -1075,6 +1124,7 @@ internal void RestartServerAndWaitForRecovery(AutorecoveringConnection conn)
             Wait(rl);
         }
 
+        /*
         internal void TestDelayedBasicAckNackAfterChannelRecovery(TestBasicConsumer1 cons, ManualResetEventSlim latch)
         {
             string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
@@ -1096,6 +1146,7 @@ internal void TestDelayedBasicAckNackAfterChannelRecovery(TestBasicConsumer1 con
             publishingModel.Close();
             publishingConn.Close();
         }
+        */
 
         internal void WaitForRecovery()
         {
@@ -1117,6 +1168,42 @@ internal void WaitForShutdown(IConnection conn)
             Wait(PrepareForShutdown(conn));
         }
 
+        public class LukeAckingBasicConsumer : DefaultBasicConsumer
+        {
+            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)
+            {
+                try
+                {
+                    Model.BasicAck(deliveryTag, false);
+                }
+                finally
+                {
+                    ++_counter;
+                    if (_counter >= _totalMessageCount)
+                    {
+                        _allMessagesSeenLatch.Set();
+                    }
+                }
+            }
+        }
+
         public class AckingBasicConsumer : TestBasicConsumer1
         {
             public AckingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)