Refactor concurrent Subscription iteration test, join created threads

Author: michaelklishin

projects/client/Unit/src/unit/TestMessagePatternsSubscription.cs

@@ -108,7 +108,7 @@ public void TestSubscriptionNack()
             Assert.AreEqual(0, ok.MessageCount);
         }
 
-        private class SubscriptionDrainer
+        protected class SubscriptionDrainer
         {
             protected Subscription m_subscription;
             protected bool m_ack;
@@ -132,23 +132,25 @@ public void Drain()
                 #pragma warning disable 0168
                 try
                 {
-                    while(!shouldStop)
+                    for(int i = 0; i < 1000; i++)
                     {
                         BasicDeliverEventArgs ea = m_subscription.Next();
                         if(ea != null)
                         {
                             Assert.That(ea, Is.TypeOf(typeof(BasicDeliverEventArgs)));
                             PostProcess();
                         } else
-                          {
-                              shouldStop = true;
-                          }
+                        {
+                            break;
+                        }
                     }
                 } catch (AlreadyClosedException ace)
                 {
-                    shouldStop = true;
+                    // expected
                 }
                 #pragma warning restore
+
+                m_subscription.Close();
             }
 
             protected void PostProcess()
@@ -166,24 +168,16 @@ protected void PostProcess()
         [Test]
         public void TestConcurrentIterationAndAck()
         {
-            IDictionary<string, object> args = new Dictionary<string, object>
-            {
-                {"x-message-ttl", 5000}
-            };
-            string q = Model.QueueDeclare("", false, true, false, args);
-            Subscription sub = new Subscription(Model, q, false);
-
-            PreparedQueue(q);
-            for (int i = 0; i < 10; i++)
-            {
-                SubscriptionDrainer drainer = new SubscriptionDrainer(sub, true);
-                Thread t = new Thread(drainer.Drain);
-                t.Start();
-            }
+            TestConcurrentIterationWithDrainer(true);
         }
 
         [Test]
         public void TestConcurrentIterationAndNack()
+        {
+            TestConcurrentIterationWithDrainer(false);
+        }
+
+        protected void TestConcurrentIterationWithDrainer(bool ack)
         {
             IDictionary<string, object> args = new Dictionary<string, object>
             {
@@ -193,12 +187,20 @@ public void TestConcurrentIterationAndNack()
             Subscription sub = new Subscription(Model, q, false);
 
             PreparedQueue(q);
+
+            List<Thread> ts = new List<Thread>();
             for (int i = 0; i < 10; i++)
             {
-                SubscriptionDrainer drainer = new SubscriptionDrainer(sub, false);
+                SubscriptionDrainer drainer = new SubscriptionDrainer(sub, ack);
                 Thread t = new Thread(drainer.Drain);
+                ts.Add(t);
                 t.Start();
             }
+
+            foreach(Thread t in ts)
+            {
+                t.Join();
+            }
         }
 
         private void PreparedQueue(string q)