locking around access to the unacked out buffer

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

3 files changed, 10 insertions, 3 deletions

diff --git a/cpp/src/qpid/framing/SessionState.cpp b/cpp/src/qpid/framing/SessionState.cpp
index 8056f4a523..8e5c9d34fc 100644
--- a/cpp/src/qpid/framing/SessionState.cpp
+++ b/cpp/src/qpid/framing/SessionState.cpp
@@ -80,8 +80,10 @@ boost::optional<SequenceNumber> SessionState::received(const AMQFrame& f) {
 bool SessionState::sent(const AMQFrame& f) {
     if (isSessionCommand(f))
         return false;
-    if (resumable)
+    if (resumable) {
+        sys::Mutex::ScopedLock l(unackedLock);
         unackedOut.push_back(f);
+    }
     ++lastSent;
     QPID_LOG(trace, "Sent # "<< lastSent << " " << id);
     return ackInterval &&
@@ -91,6 +93,7 @@ bool SessionState::sent(const AMQFrame& f) {
 }
 
 SessionState::Replay SessionState::replay() {
+    sys::Mutex::ScopedLock l(unackedLock);
     Replay r(unackedOut.size());
     std::copy(unackedOut.begin(), unackedOut.end(), r.begin());
     return r;
@@ -102,8 +105,10 @@ void SessionState::receivedAck(SequenceNumber acked) {
      if (lastSent < acked)
         throw InvalidArgumentException("Invalid sequence number in ack");
     size_t keep = lastSent - acked;
-    if (keep < unackedOut.size()) 
+    if (keep < unackedOut.size()) {
+        sys::Mutex::ScopedLock l(unackedLock);
         unackedOut.erase(unackedOut.begin(), unackedOut.end()-keep);
+    }
     solicitAckAt = std::max(solicitAckAt, SequenceNumber(acked+ackInterval));
 }
 
diff --git a/cpp/src/qpid/framing/SessionState.h b/cpp/src/qpid/framing/SessionState.h
index 361c960db1..7a12c2ac9a 100644
--- a/cpp/src/qpid/framing/SessionState.h
+++ b/cpp/src/qpid/framing/SessionState.h
@@ -25,6 +25,7 @@
 #include "qpid/framing/SequenceNumber.h"
 #include "qpid/framing/Uuid.h"
 #include "qpid/framing/AMQFrame.h"
+#include "qpid/sys/Mutex.h"
 
 #include <boost/optional.hpp>
 
@@ -127,6 +128,7 @@ class SessionState
     bool ackSolicited;
     bool suspending;
     bool resumable;
+    sys::Mutex unackedLock;
 };
 
 
diff --git a/cpp/src/tests/SessionState.cpp b/cpp/src/tests/SessionState.cpp
index f021861cd4..b8d0560c48 100644
--- a/cpp/src/tests/SessionState.cpp
+++ b/cpp/src/tests/SessionState.cpp
@@ -49,7 +49,7 @@ void sent(SessionState& session, const std::string& frames) {
 // Received chars as frames
 void received(SessionState& session, const std::string& frames) {
     for_each(frames.begin(), frames.end(),
-             bind(&SessionState::received, session, bind(frame, _1)));
+             bind(&SessionState::received, ref(session), bind(frame, _1)));
 }
 
 // Make a string from a ReplayRange.