diff options
Diffstat (limited to 'cpp/src')
-rw-r--r-- | cpp/src/qpid/framing/SessionState.cpp | 9 | ||||
-rw-r--r-- | cpp/src/qpid/framing/SessionState.h | 2 | ||||
-rw-r--r-- | cpp/src/tests/SessionState.cpp | 2 |
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. |