Tighten up sync-correctness in SubscriptionManager & Dispatcher.

Add a flush to SessionBase_0_10::sync() so it syncs in both directions.


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

5 files changed, 12 insertions, 10 deletions

diff --git a/cpp/src/qpid/client/Dispatcher.cpp b/cpp/src/qpid/client/Dispatcher.cpp
index 0bd0cb9d08..5028d68405 100644
--- a/cpp/src/qpid/client/Dispatcher.cpp
+++ b/cpp/src/qpid/client/Dispatcher.cpp
@@ -89,7 +89,7 @@ void Dispatcher::run()
                 }
             }
         }
-        sync(session).sync(); // Make sure all our acks are received before returning.
+        session.sync(); // Make sure all our acks are received before returning.
     }
     catch (const ClosedException&) {} //ignore it and return
     catch (const std::exception& e) {
diff --git a/cpp/src/qpid/client/SessionBase_0_10.cpp b/cpp/src/qpid/client/SessionBase_0_10.cpp
index 974acbfcf6..1fa958dcae 100644
--- a/cpp/src/qpid/client/SessionBase_0_10.cpp
+++ b/cpp/src/qpid/client/SessionBase_0_10.cpp
@@ -42,6 +42,7 @@ void SessionBase_0_10::flush()
 
 void SessionBase_0_10::sync()
 {
+    impl->sendFlush();          // Let the peer know our state.
     ExecutionSyncBody b;
     b.setSync(true);
     impl->send(b).wait(*impl);
diff --git a/cpp/src/qpid/client/SessionBase_0_10.h b/cpp/src/qpid/client/SessionBase_0_10.h
index e5d2b19327..3b54b733c8 100644
--- a/cpp/src/qpid/client/SessionBase_0_10.h
+++ b/cpp/src/qpid/client/SessionBase_0_10.h
@@ -75,9 +75,9 @@ class SessionBase_0_10 {
      */
     void close();
     
-    /** Synchronize the session: sync() waits until all commands
-     * issued on this session have been completed. It is equivalent to
-     * calling Session::executionSync()
+    /**
+     * Synchronize the session: sync() waits until all commands issued
+     * on this session so far have been completed by the broker.
      *
      * Note sync() is always synchronous, even on an AsyncSession object
      * because that's almost always what you want. You can call
diff --git a/cpp/src/qpid/client/SubscriptionManager.cpp b/cpp/src/qpid/client/SubscriptionManager.cpp
index 6036f153f6..3fa75a54ac 100644
--- a/cpp/src/qpid/client/SubscriptionManager.cpp
+++ b/cpp/src/qpid/client/SubscriptionManager.cpp
@@ -42,7 +42,7 @@ SubscriptionManager::SubscriptionManager(const Session& s)
 void SubscriptionManager::subscribeInternal(
     const std::string& q, const std::string& dest)
 {
-    async(session).messageSubscribe( // setFlowControl will sync.
+    session.messageSubscribe( 
         arg::queue=q, arg::destination=dest,
         arg::acceptMode=acceptMode, arg::acquireMode=acquireMode);
     setFlowControl(dest, messages, bytes, window); 
@@ -68,9 +68,10 @@ void SubscriptionManager::subscribe(
 void SubscriptionManager::setFlowControl(
     const std::string& dest, uint32_t messages,  uint32_t bytes, bool window)
 {
-    async(session).messageSetFlowMode(dest, window); 
-    async(session).messageFlow(dest, 0, messages); 
-    session.messageFlow(dest, 1, bytes); // Only need one sync
+    session.messageSetFlowMode(dest, window); 
+    session.messageFlow(dest, 0, messages); 
+    session.messageFlow(dest, 1, bytes);
+    session.sync();
 }
 
 void SubscriptionManager::setFlowControl(
@@ -91,8 +92,8 @@ AckPolicy& SubscriptionManager::getAckPolicy() { return autoAck; }
 
 void SubscriptionManager::cancel(const std::string dest)
 {
+    sync(session).messageCancel(dest);
     dispatcher.cancel(dest);
-    session.messageCancel(dest);
 }
 
 void SubscriptionManager::setAutoStop(bool set) { autoStop=set; }
diff --git a/cpp/src/qpid/client/SubscriptionManager.h b/cpp/src/qpid/client/SubscriptionManager.h
index 4ff962f67b..1a00e1beeb 100644
--- a/cpp/src/qpid/client/SubscriptionManager.h
+++ b/cpp/src/qpid/client/SubscriptionManager.h
@@ -48,7 +48,7 @@ class SubscriptionManager : public sys::Runnable
     void subscribeInternal(const std::string& q, const std::string& dest);
     
     qpid::client::Dispatcher dispatcher;
-    qpid::client::Session session;
+    qpid::client::AsyncSession session;
     uint32_t messages;
     uint32_t bytes;
     bool window;