Increased default flush interval to 1MB, send spontaneous known-completed at the flush interval.

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

1 files changed, 19 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/SessionState.cpp b/qpid/cpp/src/qpid/SessionState.cpp
index 4b4e79801f..ce8f45d085 100644
--- a/qpid/cpp/src/qpid/SessionState.cpp
+++ b/qpid/cpp/src/qpid/SessionState.cpp
@@ -90,6 +90,10 @@ bool SessionPoint::operator==(const SessionPoint& x) const {
 }
 
 
+SessionState::SendState::SendState() : unflushedSize(), replaySize(), bytesSinceKnownCompleted() {}
+
+SessionState::ReceiveState::ReceiveState() {}
+
 SessionPoint SessionState::senderGetCommandPoint() { return sender.sendPoint; }
 SequenceSet  SessionState::senderGetIncomplete() const { return sender.incomplete; }
 SessionPoint SessionState::senderGetReplayPoint() const { return sender.replayPoint; }
@@ -112,6 +116,7 @@ void SessionState::senderRecord(const AMQFrame& f) {
     stateful = true;
     if (timeout) sender.replayList.push_back(f);
     sender.unflushedSize += f.size();
+    sender.bytesSinceKnownCompleted += f.size();
     sender.replaySize += f.size();
     sender.incomplete += sender.sendPoint.command;
     sender.sendPoint.advance(f);
@@ -129,6 +134,18 @@ void SessionState::senderRecordFlush() {
     sender.unflushedSize = 0;
 }
 
+bool SessionState::senderNeedKnownCompleted() const {
+    // FIXME aconway 2008-06-04: this is unpleasant - replayFlushLimit == 0
+    // means never send spontaneous flush, but sends a knownCompleted for
+    // every completed. Need separate configuration?
+    // 
+    return sender.bytesSinceKnownCompleted >= config.replayFlushLimit;
+}
+
+void SessionState::senderRecordKnownCompleted() {
+    sender.bytesSinceKnownCompleted = 0;
+}
+
 void SessionState::senderConfirmed(const SessionPoint& confirmed) {
     if (confirmed > sender.sendPoint)
         throw InvalidArgumentException(QPID_MSG(getId() << "Confirmed commands not yet sent."));
@@ -213,7 +230,8 @@ SequenceNumber SessionState::receiverGetCurrent() const {
 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()
+SessionState::SessionState(const SessionId& i, const Configuration& c)
+  : id(i), timeout(), config(c), stateful()
 {
     QPID_LOG(debug, "SessionState::SessionState " << id << ": " << this);
 }