svn commit: r593112 - in /incubator/qpid/trunk/qpid/cpp/src/qpid/broker: Queue.cpp Queue.h

[gsim]

Author: gsim
Date: Thu Nov  8 02:08:26 2007
New Revision: 593112

URL: http://svn.apache.org/viewvc?rev=593112&view=rev
Log:
Ensure browsers are always serviced on the serializers dispatch thread to avoid 
concurrent servicing threads interfering with each other.


Modified:
    incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
    incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=593112&r1=593111&r2=593112&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Thu Nov  8 02:08:26 
2007
@@ -138,7 +138,8 @@
     if (!c || c->preAcquires()) {
         serializer.execute(dispatchCallback);
     } else {
-        serviceBrowser(c);
+        DispatchFunctor f(*this, c);
+        serializer.execute(f);
     }
 }
 

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=593112&r1=593111&r2=593112&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Thu Nov  8 02:08:26 
2007
@@ -66,11 +66,18 @@
             struct DispatchFunctor 
             {
                 Queue& queue;
+                Consumer::ptr consumer;
                 DispatchCompletion* sync;
+
                 DispatchFunctor(Queue& q, DispatchCompletion* s = 0) : 
queue(q), sync(s) {}
+                DispatchFunctor(Queue& q, Consumer::ptr c, DispatchCompletion* 
s = 0) : queue(q), consumer(c), sync(s) {}
                 void operator()()
                 {
-                    queue.dispatch(); 
+                    if (consumer && !consumer->preAcquires()) {
+                        queue.serviceBrowser(consumer);                        
+                    }else{
+                        queue.dispatch(); 
+                    }
                     if (sync) sync->completed();
                 }
             };