Fixed issue with producer flow control in a cluster.

Producer flow control uses a Timer and other clock-based calculations to send flow control commands.
These commands are not predictably ordered from the clusters point of view.

Added getClusterOrderProxy() to SessionState. In a cluster it returns
a proxy that defers sending a command to the client until it is
multicast to the cluster.  In a stand alone broker it is just the
normal proxy. Updated producer flow control to use this proxy.

Cluster flow control is turned off in shadow connections. Only the
directly connected node does flow control calculations and multicasts
the commands to send. All nodes sending of the commands thru SessionState
to ensure consistent session state (e.g. command numbering.)


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

1 files changed, 10 insertions, 1 deletions

diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h
index c435a741f8..b64461eb86 100644
--- a/cpp/src/qpid/broker/SessionState.h
+++ b/cpp/src/qpid/broker/SessionState.h
@@ -125,6 +125,15 @@ class SessionState : public qpid::SessionState,
 
     void sendAcceptAndCompletion();
 
+    /**
+     * If commands are sent based on the local time (e.g. in timers), they don't have
+     * a well-defined ordering across cluster nodes.
+     * This proxy is for sending such commands. In a clustered broker it will take steps
+     * to synchronize command order across the cluster. In a stand-alone broker
+     * it is just a synonym for getProxy()
+     */  
+    framing::AMQP_ClientProxy& getClusterOrderProxy();
+    
     Broker& broker;
     SessionHandler* handler;
     sys::AbsTime expiry;        // Used by SessionManager.
@@ -138,7 +147,7 @@ class SessionState : public qpid::SessionState,
 
     // State used for producer flow control (rate limited)
     qpid::sys::Mutex rateLock;
-    RateFlowcontrol* rateFlowcontrol;
+    boost::scoped_ptr<RateFlowcontrol> rateFlowcontrol;
     boost::intrusive_ptr<TimerTask> flowControlTimer;
 
     friend class SessionManager;