You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2011/06/24 16:42:23 UTC
svn commit: r1139334 - in /qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10:
AcceptTracker.cpp AcceptTracker.h
Author: gsim
Date: Fri Jun 24 14:42:23 2011
New Revision: 1139334
URL: http://svn.apache.org/viewvc?rev=1139334&view=rev
Log:
QPID-3321: flush occasionally to keep the list of pending accepts from growing too large
Modified:
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp?rev=1139334&r1=1139333&r2=1139334&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp Fri Jun 24 14:42:23 2011
@@ -70,6 +70,18 @@ void AcceptTracker::delivered(const std:
destinationState[destination].unaccepted.add(id);
}
+namespace
+{
+const size_t FLUSH_FREQUENCY = 1024;
+}
+
+void AcceptTracker::addToPending(qpid::client::AsyncSession& session, const Record& record)
+{
+ pending.push_back(record);
+ if (pending.size() > FLUSH_FREQUENCY) session.flush();
+}
+
+
void AcceptTracker::accept(qpid::client::AsyncSession& session)
{
for (StateMap::iterator i = destinationState.begin(); i != destinationState.end(); ++i) {
@@ -78,7 +90,7 @@ void AcceptTracker::accept(qpid::client:
Record record;
record.status = session.messageAccept(aggregateState.unaccepted);
record.accepted = aggregateState.unaccepted;
- pending.push_back(record);
+ addToPending(session, record);
aggregateState.accept();
}
@@ -90,7 +102,7 @@ void AcceptTracker::accept(qpid::framing
Record record;
record.accepted = aggregateState.accept(id, cumulative);
record.status = session.messageAccept(record.accepted);
- pending.push_back(record);
+ addToPending(session, record);
}
void AcceptTracker::release(qpid::client::AsyncSession& session)
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h?rev=1139334&r1=1139333&r2=1139334&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h Fri Jun 24 14:42:23 2011
@@ -79,6 +79,7 @@ class AcceptTracker
StateMap destinationState;
Records pending;
+ void addToPending(qpid::client::AsyncSession&, const Record&);
void checkPending();
void completed(qpid::framing::SequenceSet&);
};
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org