RhinoQueuesTransport waits for current message to finish processing b…

…efore disposing the queue manager so messages can still be enqueued by consumers.
ozanyurt · Sep 14, 2012 · 74fa750 · 74fa750
1 parent 4ee8754
commit 74fa750
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 2 deletions.
diff --git a/Rhino.ServiceBus.RhinoQueues/RhinoQueues/RhinoQueuesTransport.cs b/Rhino.ServiceBus.RhinoQueues/RhinoQueues/RhinoQueuesTransport.cs
@@ -34,6 +34,7 @@ public class RhinoQueuesTransport : ITransport
         private QueueManager queueManager;
         private readonly Thread[] threads;
         private readonly string queueName;
+        private volatile int currentlyProccessingCount;
         private volatile bool shouldContinue;
         private bool haveStarted;
         private readonly IsolationLevel queueIsolationLevel;
@@ -84,17 +85,19 @@ public void Dispose()
             shouldContinue = false;
             logger.DebugFormat("Stopping transport for {0}", endpoint);
 
+            while (currentlyProccessingCount > 0)
+                Thread.Sleep(TimeSpan.FromSeconds(1));
+
             if (timeout != null)
                 timeout.Dispose();
             DisposeQueueManager();
 
             if (!haveStarted)
                 return;
-
             foreach (var thread in threads)
             {
                 thread.Join();
-            }
+            } 
         }
 
         private void DisposeQueueManager()
@@ -364,6 +367,8 @@ private void ProcessMessage(Message message, TransactionScope tx, Func<CurrentMe
             try
             {
                 //deserialization errors do not count for module events
+#pragma warning disable 420
+                Interlocked.Increment(ref currentlyProccessingCount);
                 object[] messages = DeserializeMessages(message);
                 try
                 {
@@ -404,6 +409,8 @@ private void ProcessMessage(Message message, TransactionScope tx, Func<CurrentMe
                                                                               MessageProcessingFailure, currentMessageInformation);
                 messageHandlingCompletion.HandleMessageCompletion();
                 currentMessageInformation = null;
+                Interlocked.Decrement(ref currentlyProccessingCount);
+#pragma warning restore 420
             }
         }
 

diff --git a/Rhino.ServiceBus.Tests/CanCustomizeOutgoingMessagesWithMsmq.cs b/Rhino.ServiceBus.Tests/CanCustomizeOutgoingMessagesWithMsmq.cs
@@ -108,6 +108,7 @@ public void TimeToReachQueue_is_set_when_MaxAttempts_is_one()
         {
             using (var container = new WindsorContainer())
             {
+                MaxAttemptCustomizer.MaxAttempts = 1;
                 container.Register(Component.For<ICustomizeOutgoingMessages>()
                     .ImplementedBy<MaxAttemptCustomizer>()
                     .LifeStyle.Is(LifestyleType.Transient));