Merge branch 'hotfix-2.0.5'

Particular · Dec 5, 2014 · a890428 · a890428
2 parents 38a43b8 + 23478e5
commit a890428
Showing 1 changed file with 24 additions and 21 deletions.
diff --git a/src/NServiceBus.RabbitMQ/RabbitMqDequeueStrategy.cs b/src/NServiceBus.RabbitMQ/RabbitMqDequeueStrategy.cs
@@ -50,7 +50,7 @@ public void Init(Address address, TransactionSettings transactionSettings, Func<
             this.endProcessMessage = endProcessMessage;
             workQueue = address.Queue;
 
-            autoAck = !transactionSettings.IsTransactional;
+            noAck = !transactionSettings.IsTransactional;
 
             if (purgeOnStartup)
             {
@@ -146,10 +146,13 @@ void StartConsumer(string queue)
 
         void ConsumeMessages(object state)
         {
-            try
+            if (!tracksRunningThreads.Wait(TimeSpan.FromSeconds(1)))
             {
-                tracksRunningThreads.Wait();
+                return;
+            }
 
+            try
+            {
                 var parameters = (ConsumeParams)state;
                 var connection = connectionManager.GetConsumeConnection();
 
@@ -159,7 +162,7 @@ void ConsumeMessages(object state)
 
                     var consumer = new QueueingBasicConsumer(channel);
 
-                    channel.BasicConsume(parameters.Queue, autoAck, consumer);
+                    channel.BasicConsume(parameters.Queue, noAck, consumer);
 
                     circuitBreaker.Success();
 
@@ -196,7 +199,7 @@ void ConsumeMessages(object state)
                                 messageProcessedOk = tryProcessMessage(transportMessage);
                             }
 
-                            if (!autoAck)
+                            if (!noAck)
                             {
                                 if (messageProcessedOk)
                                 {
@@ -212,7 +215,7 @@ void ConsumeMessages(object state)
                         {
                             exception = ex;
 
-                            if (!autoAck)
+                            if (!noAck)
                             {
                                 channel.BasicReject(message.DeliveryTag, true);
                             }
@@ -224,6 +227,18 @@ void ConsumeMessages(object state)
                     }
                 }
             }
+            catch (EndOfStreamException)
+            {
+                // If no items are present and the queue is in a closed
+                // state, or if at any time while waiting the queue
+                // transitions to a closed state (by a call to Close()), this
+                // method will throw EndOfStreamException.
+
+                // We need to put a delay here otherwise we end-up doing a tight loop that causes
+                // CPU spikes
+                Thread.Sleep(1000);
+                throw;
+            }
             catch (IOException)
             {
                 //Unable to write data to the transport connection: An existing connection was forcibly closed by the remote host.
@@ -237,21 +252,9 @@ void ConsumeMessages(object state)
 
         static BasicDeliverEventArgs DequeueMessage(QueueingBasicConsumer consumer)
         {
-            BasicDeliverEventArgs rawMessage = null;
-
-            var messageDequeued = false;
+            BasicDeliverEventArgs rawMessage;
 
-            try
-            {
-                messageDequeued = consumer.Queue.Dequeue(1000, out rawMessage);
-            }
-            catch (EndOfStreamException)
-            {
-                // If no items are present and the queue is in a closed
-                // state, or if at any time while waiting the queue
-                // transitions to a closed state (by a call to Close()), this
-                // method will throw EndOfStreamException.
-            }
+            var messageDequeued = consumer.Queue.Dequeue(1000, out rawMessage);
 
             if (!messageDequeued)
             {
@@ -273,7 +276,7 @@ void Purge()
 
         RepeatedFailuresOverTimeCircuitBreaker circuitBreaker;
 
-        bool autoAck;
+        bool noAck;
         SemaphoreSlim tracksRunningThreads;
         Action<TransportMessage, Exception> endProcessMessage;
         CancellationTokenSource tokenSource;