Ensure we wait for all messages to be consumed

Author: lukebakken

projects/Unit/TestBasicPublish.cs

@@ -30,14 +30,12 @@ public void TestBasicRoundtripArray()
                 };
                 string tag = m.BasicConsume(q.QueueName, true, consumer);
 
-
                 m.BasicPublish("", q.QueueName, bp, sendBody);
                 bool waitResFalse = are.WaitOne(2000);
                 m.BasicCancel(tag);
 
-
                 Assert.IsTrue(waitResFalse);
-                Assert.AreEqual(sendBody, consumeBody);
+                CollectionAssert.AreEqual(sendBody, consumeBody);
             }
         }
 
@@ -62,14 +60,12 @@ public void TestBasicRoundtripReadOnlyMemory()
                 };
                 string tag = m.BasicConsume(q.QueueName, true, consumer);
 
-
                 m.BasicPublish("", q.QueueName, bp, new ReadOnlyMemory<byte>(sendBody));
                 bool waitResFalse = are.WaitOne(2000);
                 m.BasicCancel(tag);
 
-
                 Assert.IsTrue(waitResFalse);
-                Assert.AreEqual(sendBody, consumeBody);
+                CollectionAssert.AreEqual(sendBody, consumeBody);
             }
         }
     }

projects/Unit/TestFloodPublishing.cs

@@ -103,6 +103,7 @@ public void TestMultithreadFloodPublishing()
             byte[] sendBody = Encoding.UTF8.GetBytes(message);
             int publishCount = 4096;
             int receivedCount = 0;
+            AutoResetEvent autoResetEvent = new AutoResetEvent(false);
 
             var cf = new ConnectionFactory()
             {
@@ -139,11 +140,17 @@ public void TestMultithreadFloodPublishing()
                         string receivedMessage = Encoding.UTF8.GetString(a.Body.ToArray());
                         Assert.AreEqual(message, receivedMessage);
                         Interlocked.Increment(ref receivedCount);
+                        if (receivedCount == publishCount)
+                        {
+                            autoResetEvent.Set();
+                        }
                     };
                     consumerModel.BasicConsume(queueName, true, consumer);
                     Assert.IsTrue(pub.Wait(_tenSeconds));
-                    Assert.AreEqual(publishCount, receivedCount);
+                    Assert.IsTrue(autoResetEvent.WaitOne(_tenSeconds));
                 }
+
+                Assert.AreEqual(publishCount, receivedCount);
             }
         }
     }

projects/Unit/TestRecoverAfterCancel.cs

@@ -96,7 +96,7 @@ public void TestRecoverAfterCancel_()
             Channel.BasicConsume(Queue, false, Consumer2);
             BasicDeliverEventArgs Event2 = EventQueue2.Dequeue();
 
-            Assert.AreEqual(Event.Body, Event2.Body);
+            CollectionAssert.AreEqual(Event.Body.ToArray(), Event2.Body.ToArray());
             Assert.IsFalse(Event.Redelivered);
             Assert.IsTrue(Event2.Redelivered);
         }