1 files changed, 24 insertions, 27 deletions

diff --git a/cpp/src/qpid/client/Bounds.cpp b/cpp/src/qpid/client/Bounds.cpp
index 1df21db941..aac18022bc 100644
--- a/cpp/src/qpid/client/Bounds.cpp
+++ b/cpp/src/qpid/client/Bounds.cpp
@@ -1,49 +1,40 @@
 #include "Bounds.h"
 
 #include "qpid/log/Statement.h"
+#include "qpid/sys/Waitable.h"
 
 namespace qpid {
 namespace client {
 
-using sys::Monitor;
+using sys::Waitable;
 
 Bounds::Bounds(size_t maxSize) : max(maxSize), current(0) {}
 
-bool Bounds::expand(size_t sizeRequired, bool block) 
-{
-    if (max) {
-        Monitor::ScopedLock l(lock);
-        current += sizeRequired;
-        if (block) {
-            while (current > max) {
-                QPID_LOG(debug, "Waiting for bounds: " << *this); 
-                lock.wait();
-            }
-            QPID_LOG(debug, "Bounds ok: " << *this);
-        }
-        return current <= max;
-    } else {
-        return true;
+bool Bounds::expand(size_t sizeRequired, bool block) {
+    if (!max) return true;
+    Waitable::ScopedLock l(lock);
+    current += sizeRequired;
+    if (block) {
+        Waitable::ScopedWait w(lock);
+        while (current > max) 
+            lock.wait();
     }
+    return current <= max;
 }
 
-void Bounds::reduce(size_t size)
-{
+void Bounds::reduce(size_t size) {
     if (!max || size == 0) return;
-    Monitor::ScopedLock l(lock);
+    Waitable::ScopedLock l(lock);
     if (current == 0) return;
-    bool needNotify = current > max;
     current -= std::min(size, current);
-    if (needNotify && current < max) {
-        //todo: notify one at a time, but ensure that all threads are
-        //eventually notified
-        lock.notifyAll();
+    if (current < max && lock.hasWaiters()) {
+        assert(lock.hasWaiters() == 1);
+        lock.notify();
     }
 }
 
-size_t Bounds::getCurrentSize()
-{
-    Monitor::ScopedLock l(lock);
+size_t Bounds::getCurrentSize() {
+    Waitable::ScopedLock l(lock);
     return current;
 }
 
@@ -52,4 +43,10 @@ std::ostream& operator<<(std::ostream& out, const Bounds& bounds) {
     return out;
 }
 
+void Bounds::setException(const sys::ExceptionHolder& e) {
+    Waitable::ScopedLock l(lock);    
+    lock.setException(e);
+    lock.waitWaiters();         // Wait for waiting threads to exit.
+}
+
 }} // namespace qpid::client