QPID-1264: initial fix for fanout, direct and headers exchanges (fix for remaining types to follow)

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@693053 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 11 insertions, 31 deletions

diff --git a/cpp/src/qpid/broker/FanOutExchange.cpp b/cpp/src/qpid/broker/FanOutExchange.cpp
index 373e9ab1cc..019c943ca1 100644
--- a/cpp/src/qpid/broker/FanOutExchange.cpp
+++ b/cpp/src/qpid/broker/FanOutExchange.cpp
@@ -40,18 +40,10 @@ FanOutExchange::FanOutExchange(const std::string& _name, bool _durable,
         mgmtExchange->set_type (typeName);
 }
 
-bool FanOutExchange::bind(Queue::shared_ptr queue, const string& /*routingKey*/, const FieldTable* /*args*/){
-    RWlock::ScopedWlock locker(lock);
-    std::vector<Binding::shared_ptr>::iterator i;
-
-    // Add if not already present.
-    for (i = bindings.begin (); i != bindings.end(); i++)
-        if ((*i)->queue == queue)
-            break;
-
-    if (i == bindings.end()) {
-        Binding::shared_ptr binding (new Binding ("", queue, this));
-        bindings.push_back(binding);
+bool FanOutExchange::bind(Queue::shared_ptr queue, const string& /*key*/, const FieldTable* /*args*/)
+{
+    Binding::shared_ptr binding (new Binding ("", queue, this));
+    if (bindings.add_unless(binding, MatchQueue(queue))) {
         if (mgmtExchange != 0) {
             mgmtExchange->inc_bindingCount();
             ((management::Queue*) queue->GetManagementObject())->inc_bindingCount();
@@ -62,16 +54,9 @@ bool FanOutExchange::bind(Queue::shared_ptr queue, const string& /*routingKey*/,
     }
 }
 
-bool FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*/, const FieldTable* /*args*/){
-    RWlock::ScopedWlock locker(lock);
-    std::vector<Binding::shared_ptr>::iterator i;
-
-    for (i = bindings.begin (); i != bindings.end(); i++)
-        if ((*i)->queue == queue)
-            break;
-
-    if (i != bindings.end()) {
-        bindings.erase(i);
+bool FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*key*/, const FieldTable* /*args*/)
+{
+    if (bindings.remove_if(MatchQueue(queue))) {
         if (mgmtExchange != 0) {
             mgmtExchange->dec_bindingCount();
             ((management::Queue*) queue->GetManagementObject())->dec_bindingCount();
@@ -83,10 +68,10 @@ bool FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*
 }
 
 void FanOutExchange::route(Deliverable& msg, const string& /*routingKey*/, const FieldTable* /*args*/){
-    RWlock::ScopedRlock locker(lock);
     uint32_t count(0);
 
-    for(std::vector<Binding::shared_ptr>::iterator i = bindings.begin(); i != bindings.end(); ++i, count++){
+    BindingsArray::ConstPtr p = bindings.snapshot();
+    for(std::vector<Binding::shared_ptr>::const_iterator i = p->begin(); i != p->end(); ++i, count++){
         msg.deliverTo((*i)->queue);
         if ((*i)->mgmtBinding != 0)
             (*i)->mgmtBinding->inc_msgMatched ();
@@ -111,13 +96,8 @@ void FanOutExchange::route(Deliverable& msg, const string& /*routingKey*/, const
 
 bool FanOutExchange::isBound(Queue::shared_ptr queue, const string* const, const FieldTable* const)
 {
-    std::vector<Binding::shared_ptr>::iterator i;
-
-    for (i = bindings.begin (); i != bindings.end(); i++)
-        if ((*i)->queue == queue)
-            break;
-
-    return i != bindings.end();
+    BindingsArray::ConstPtr ptr = bindings.snapshot();
+    return ptr && std::find_if(ptr->begin(), ptr->end(), MatchQueue(queue)) != ptr->end();
 }