QPID-1478: ensure concurrent publishers work correctly (as well as reported assertion, the test uncovered a potential deadlock due to bounds being expanded before frames were added to queue).

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

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);
     }
 }