summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/client
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/client')
-rw-r--r--cpp/src/qpid/client/Bounds.cpp3
-rw-r--r--cpp/src/qpid/client/Connector.cpp4
-rw-r--r--cpp/src/qpid/client/SessionImpl.cpp2
3 files changed, 5 insertions, 4 deletions
diff --git a/cpp/src/qpid/client/Bounds.cpp b/cpp/src/qpid/client/Bounds.cpp
index bbc08edc5e..e2ddb5dfec 100644
--- a/cpp/src/qpid/client/Bounds.cpp
+++ b/cpp/src/qpid/client/Bounds.cpp
@@ -48,8 +48,7 @@ void Bounds::reduce(size_t size) {
if (current == 0) return;
current -= std::min(size, current);
if (current < max && lock.hasWaiters()) {
- assert(lock.hasWaiters() == 1);
- lock.notify();
+ lock.notifyAll();
}
}
diff --git a/cpp/src/qpid/client/Connector.cpp b/cpp/src/qpid/client/Connector.cpp
index ea32c4e097..724d464932 100644
--- a/cpp/src/qpid/client/Connector.cpp
+++ b/cpp/src/qpid/client/Connector.cpp
@@ -294,7 +294,9 @@ void TCPConnector::Writer::init(std::string id, sys::AsynchIO* a) {
void TCPConnector::Writer::handle(framing::AMQFrame& frame) {
Mutex::ScopedLock l(lock);
frames.push_back(frame);
- if (frame.getEof()) {//or if we already have a buffers worth
+ //only try to write if this is the end of a frameset or if we
+ //already have a buffers worth of data
+ if (frame.getEof() || (bounds && bounds->getCurrentSize() >= maxFrameSize)) {
lastEof = frames.size();
aio->notifyPendingWrite();
}
diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index cf71d4f4a5..ab8c1bddb8 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -451,8 +451,8 @@ void SessionImpl::sendFrame(AMQFrame& frame, bool canBlock)
{
boost::shared_ptr<ConnectionImpl> c = connectionWeak.lock();
if (c) {
- c->expand(frame.encodedSize(), canBlock);
channel.handle(frame);
+ c->expand(frame.encodedSize(), canBlock);
}
}