Merge branch 'hotfix-2.1.2'

Author: andreasohlund

src/NServiceBus.RabbitMQ.Tests/When_stopping_endpoint.cs

@@ -22,10 +22,10 @@ public void Should__gracefully_shutdown()
             var address = Address.Parse(ReceiverQueue);
 
             Parallel.For(0, 2000, i =>
-                sender.Send(new TransportMessage(), new SendOptions(address)));
+                sender.Send(new TransportMessage(){Body = new byte[1]}, new SendOptions(address)));
 
             dequeueStrategy.Start(50);
-            Thread.Sleep(10);
+            Thread.Sleep(1000);
             dequeueStrategy.Stop();
             connectionManager.Dispose();
         }

src/NServiceBus.RabbitMQ/RabbitMqDequeueStrategy.cs

@@ -36,6 +36,8 @@ public void Init(Address address, TransactionSettings transactionSettings, Func<
 
         public void Start(int maximumConcurrencyLevel)
         {
+            isStopping = false;
+
             if (receiveOptions.DefaultPrefetchCount > 0)
             {
                 actualPrefetchCount = receiveOptions.DefaultPrefetchCount;
@@ -79,16 +81,26 @@ public void Start(int maximumConcurrencyLevel)
         /// </summary>
         public void Stop()
         {
+            if (isStopping)
+            {
+                return;
+            }
+            
+            isStopping = true;
+
             if (tokenSource == null)
             {
                 return;
             }
 
             tokenSource.Cancel();
-            DrainStopSemaphore();
+
+            WaitForThreadsToStop();
+
+            tokenSource = null;
         }
 
-        void DrainStopSemaphore()
+        void WaitForThreadsToStop()
         {
             for (var index = 0; index < actualConcurrencyLevel; index++)
             {
@@ -114,6 +126,7 @@ void StartConsumer(string queue)
                 {
                     t.Exception.Handle(ex =>
                     {
+                        Logger.Error("Failed to receive messages from " + queue,t.Exception);
                         circuitBreaker.Failure(ex);
                         return true;
                     });
@@ -224,6 +237,10 @@ void ConsumeMessages(object state)
             {
                 //Unable to write data to the transport connection: An existing connection was forcibly closed by the remote host.
                 //This exception is expected because we are shutting down!
+                if (!isStopping)
+                {
+                    throw;
+                }
             }
             finally
             {
@@ -270,6 +287,7 @@ void Purge()
         readonly IManageRabbitMqConnections connectionManager;
         readonly ReceiveOptions receiveOptions;
         ushort actualPrefetchCount;
+        bool isStopping;
 
 
         class ConsumeParams