QPID-1711: Ensure the session state between the two peers in an inter-broker bridging session are kept in sync.

(Also made changes to cancellation to ensure that the commands are only issued on the io thread of the connection)



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

1 files changed, 28 insertions, 15 deletions

diff --git a/cpp/src/qpid/broker/Link.cpp b/cpp/src/qpid/broker/Link.cpp
index e36635831b..dd1a1fa0b4 100644
--- a/cpp/src/qpid/broker/Link.cpp
+++ b/cpp/src/qpid/broker/Link.cpp
@@ -158,7 +158,7 @@ void Link::closed (int, std::string text)
     }
 
     for (Bridges::iterator i = active.begin(); i != active.end(); i++) {
-        (*i)->cancel();
+        (*i)->closed();
         created.push_back(*i);
     }
     active.clear();
@@ -217,21 +217,27 @@ void Link::add(Bridge::shared_ptr bridge)
 
 void Link::cancel(Bridge::shared_ptr bridge)
 {
-    Mutex::ScopedLock mutex(lock);
-
-    for (Bridges::iterator i = created.begin(); i != created.end(); i++) {
-        if ((*i).get() == bridge.get()) {
-            created.erase(i);
-            break;
+    {
+        Mutex::ScopedLock mutex(lock);
+        
+        for (Bridges::iterator i = created.begin(); i != created.end(); i++) {
+            if ((*i).get() == bridge.get()) {
+                created.erase(i);
+                break;
+            }
         }
-    }
-    for (Bridges::iterator i = active.begin(); i != active.end(); i++) {
-        if ((*i).get() == bridge.get()) {
-            bridge->cancel();
-            active.erase(i);
-            break;
+        for (Bridges::iterator i = active.begin(); i != active.end(); i++) {
+            if ((*i).get() == bridge.get()) {
+                cancellations.push_back(bridge);
+                bridge->closed();
+                active.erase(i);
+                break;
+            }
         }
     }
+    if (!cancellations.empty()) {
+        connection->requestIOProcessing (boost::bind(&Link::ioThreadProcessing, this));
+    }
 }
 
 void Link::ioThreadProcessing()
@@ -242,7 +248,7 @@ void Link::ioThreadProcessing()
         return;
     QPID_LOG(debug, "Link::ioThreadProcessing()");
 
-    //process any pending creates
+    //process any pending creates and/or cancellations
     if (!created.empty()) {
         for (Bridges::iterator i = created.begin(); i != created.end(); ++i) {
             active.push_back(*i);
@@ -250,6 +256,13 @@ void Link::ioThreadProcessing()
         }
         created.clear();
     }
+    if (!cancellations.empty()) {
+        for (Bridges::iterator i = cancellations.begin(); i != cancellations.end(); ++i) {
+            active.push_back(*i);
+            (*i)->cancel(*connection);
+        }
+        cancellations.clear();
+    }
 }
 
 void Link::setConnection(Connection* c)
@@ -284,7 +297,7 @@ void Link::maintenanceVisit ()
             }
         }
     }
-    else if (state == STATE_OPERATIONAL && !created.empty() && connection != 0)
+    else if (state == STATE_OPERATIONAL && (!created.empty() || !cancellations.empty()) && connection != 0)
         connection->requestIOProcessing (boost::bind(&Link::ioThreadProcessing, this));
 }