summaryrefslogtreecommitdiff
path: root/cpp
diff options
context:
space:
mode:
Diffstat (limited to 'cpp')
-rw-r--r--cpp/src/qpid/client/Dispatcher.cpp2
-rw-r--r--cpp/src/qpid/client/SessionBase_0_10.cpp1
-rw-r--r--cpp/src/qpid/client/SessionBase_0_10.h6
-rw-r--r--cpp/src/qpid/client/SubscriptionManager.cpp11
-rw-r--r--cpp/src/qpid/client/SubscriptionManager.h2
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;