QPID-1711: Ensure the session state between the two peers in an inter-broker bridging session are kept in sync.

(Also made changes to cancellation to ensure that the commands are only issued on the io thread of the connection)



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

1 files changed, 6 insertions, 1 deletions

diff --git a/cpp/src/qpid/SessionState.cpp b/cpp/src/qpid/SessionState.cpp
index ac75b5c5ff..2ea6b39f72 100644
--- a/cpp/src/qpid/SessionState.cpp
+++ b/cpp/src/qpid/SessionState.cpp
@@ -183,6 +183,7 @@ void SessionState::receiverSetCommandPoint(const SessionPoint& point) {
 }
 
 bool SessionState::receiverRecord(const AMQFrame& f) {
+    if (receiverTrackingDisabled) return true; //Very nasty hack for push bridges
     if (isControl(f)) return true; // Ignore control frames.
     stateful = true;
     receiver.expected.advance(f);
@@ -198,6 +199,7 @@ bool SessionState::receiverRecord(const AMQFrame& f) {
 }
     
 void SessionState::receiverCompleted(SequenceNumber command, bool cumulative) {
+    if (receiverTrackingDisabled) return; //Very nasty hack for push bridges
     assert(receiver.incomplete.contains(command)); // Internal error to complete command twice.
     SequenceNumber first =cumulative ? receiver.incomplete.front() : command;
     SequenceNumber last = command;
@@ -237,7 +239,7 @@ SessionState::Configuration::Configuration(size_t flush, size_t hard) :
     replayFlushLimit(flush), replayHardLimit(hard) {}
 
 SessionState::SessionState(const SessionId& i, const Configuration& c)
-    : id(i), timeout(), config(c), stateful()
+    : id(i), timeout(), config(c), stateful(), receiverTrackingDisabled(false)
 {
     QPID_LOG(debug, "SessionState::SessionState " << id << ": " << this);
 }
@@ -275,4 +277,7 @@ void SessionState::setState(
     receiver.bytesSinceKnownCompleted = 0;
 }
 
+void SessionState::disableReceiverTracking() { receiverTrackingDisabled = true; }
+void SessionState::enableReceiverTracking() { receiverTrackingDisabled = false; }
+
 } // namespace qpid