Added --cluster-read-max: max number of outstanding mcasts in CPG buffers.

Work around problems with CPG flow control.


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

1 files changed, 9 insertions, 2 deletions

diff --git a/cpp/src/qpid/cluster/Multicaster.h b/cpp/src/qpid/cluster/Multicaster.h
index a63bbc2180..ef8be8c229 100644
--- a/cpp/src/qpid/cluster/Multicaster.h
+++ b/cpp/src/qpid/cluster/Multicaster.h
@@ -46,13 +46,19 @@ class Multicaster
 {
   public:
     /** Starts in holding mode: connection data events are held, other events are mcast */
-    Multicaster(Cpg& cpg_, const boost::shared_ptr<sys::Poller>&, boost::function<void()> onError );
+    Multicaster(Cpg& cpg_,
+                size_t mcastMax, 
+                const boost::shared_ptr<sys::Poller>&,
+                boost::function<void()> onError
+    );
     void mcastControl(const framing::AMQBody& controlBody, const ConnectionId&);
     void mcastBuffer(const char*, size_t, const ConnectionId&);
     void mcast(const Event& e);
     /** End holding mode, held events are mcast */
     void release();
-
+    /** Call when events are self-delivered to manage flow control. */
+    void delivered(const Event& e);
+    
   private:
     typedef sys::PollableQueue<Event> PollableEventQueue;
     typedef std::deque<Event> PlainEventQueue;
@@ -66,6 +72,7 @@ class Multicaster
     bool holding;
     PlainEventQueue holdingQueue;
     std::vector<struct ::iovec> ioVector;
+    size_t mcastMax, pending;
 };
 }} // namespace qpid::cluster