svn commit: r784267 - in /qpid/trunk/qpid/cpp/src/qpid/broker: QueueListeners.cpp QueueListeners.h

[ac...@apache.org]

Author: aconway
Date: Fri Jun 12 20:42:16 2009
New Revision: 784267

URL: http://svn.apache.org/viewvc?rev=784267&view=rev
Log:
Revert bad optimization:  unpredictable ordering of set caused cluster failures.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp?rev=784267&r1=784266&r2=784267&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp Fri Jun 12 20:42:16 2009
@@ -27,31 +27,43 @@
 void QueueListeners::addListener(Consumer::shared_ptr c)
 {
     if (c->preAcquires()) {
-        consumers.insert(c);
+        add(consumers, c);
     } else {
-        browsers.insert(c);
+        add(browsers, c);
     }
 }
 
 void QueueListeners::removeListener(Consumer::shared_ptr c)
 {
     if (c->preAcquires()) {
-        consumers.erase(c);
+        remove(consumers, c);
     } else {
-        browsers.erase(c);
+        remove(browsers, c);
     }
 }
 
 void QueueListeners::populate(NotificationSet& set)
 {
-    if (!consumers.empty()) {
-        set.consumer = *consumers.begin();
-        consumers.erase(consumers.begin());
+    if (consumers.size()) {
+        set.consumer = consumers.front();
+        consumers.pop_front();
     } else {
         browsers.swap(set.browsers);
     }
 }
 
+void QueueListeners::add(Listeners& listeners, Consumer::shared_ptr c)
+{
+    Listeners::iterator i = std::find(listeners.begin(), listeners.end(), c);
+    if (i == listeners.end()) listeners.push_back(c);
+}
+
+void QueueListeners::remove(Listeners& listeners, Consumer::shared_ptr c)
+{
+    Listeners::iterator i = std::find(listeners.begin(), listeners.end(), c);
+    if (i != listeners.end()) listeners.erase(i);
+}
+
 void QueueListeners::NotificationSet::notify()
 {
     if (consumer) consumer->notify();

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h?rev=784267&r1=784266&r2=784267&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h Fri Jun 12 20:42:16 2009
@@ -22,7 +22,7 @@
  *
  */
 #include "Consumer.h"
-#include <set>
+#include <list>
 
 namespace qpid {
 namespace broker {
@@ -40,7 +40,7 @@
 class QueueListeners
 {
   public:
-    typedef std::set<Consumer::shared_ptr> Listeners;
+    typedef std::list<Consumer::shared_ptr> Listeners;
 
     class NotificationSet
     {
@@ -58,6 +58,10 @@
   private:
     Listeners consumers;
     Listeners browsers;
+
+    void add(Listeners&, Consumer::shared_ptr);
+    void remove(Listeners&, Consumer::shared_ptr);
+
 };
 }} // namespace qpid::broker
 



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org