Replicate connection decoder fragments to new members.

Refactoring:
 - Merge Decoder into ConnectionMap.
 - Process cluster controls in event queue thread.
 - Use counter not pointer for connection ID, avoid re-use.
 - Do all processing in event queue thread to avoid races
   (temporary pending performance measurements)


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

1 files changed, 14 insertions, 7 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index ea472a9ecf..4d358cf495 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -30,7 +30,6 @@
 #include "NoOpConnectionOutputHandler.h"
 #include "PollerDispatch.h"
 #include "Quorum.h"
-#include "Decoder.h"
 #include "PollableQueue.h"
 #include "ExpiryPolicy.h"
 
@@ -102,7 +101,10 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     size_t getWriteEstimate() { return writeEstimate; }
 
     bool isLeader() const;       // Called in deliver thread.
-    
+
+    // Called by Connection in deliver event thread with decoded connection data frames.
+    void connectionFrame(const EventFrame&); 
+
   private:
     typedef sys::Monitor::ScopedLock Lock;
 
@@ -125,7 +127,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     void brokerShutdown();
 
     // Cluster controls implement XML methods from cluster.xml.
-    // Called in deliver thread.
+    // Called in deliveredEvent thread.
     // 
     void updateRequest(const MemberId&, const std::string&, Lock&);
     void updateOffer(const MemberId& updater, uint64_t updatee, const framing::Uuid&, Lock&);
@@ -134,6 +136,10 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     void messageExpired(const MemberId&, uint64_t, Lock& l);
     void shutdown(const MemberId&, Lock&);
 
+    // Used by cluster controls.
+    void stall(Lock&);
+    void unstall(Lock&);
+
     // Handlers for pollable queues.
     void deliveredEvent(const Event&); 
     void deliveredFrame(const EventFrame&); 
@@ -141,6 +147,10 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     // Helper, called in deliver thread.
     void updateStart(const MemberId& updatee, const Url& url, Lock&);
 
+    // Called in event deliver thread to check for update status.
+    bool isUpdateComplete(const EventFrame&);
+    bool isUpdateComplete();
+
     void setReady(Lock&);
 
     void deliver( // CPG deliver callback. 
@@ -186,7 +196,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     Cpg cpg;
     const std::string name;
     Url myUrl;
-    const MemberId myId;
+    const MemberId self;
     const size_t readMax;
     const size_t writeEstimate;
     framing::Uuid clusterId;
@@ -201,9 +211,6 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     boost::shared_ptr<FailoverExchange> failoverExchange;
     Quorum quorum;
 
-    // Used only in deliverdEvent thread
-    Decoder decoder;
-
     // Used only in deliveredFrame thread
     ClusterMap::Set elders;
     boost::intrusive_ptr<ExpiryPolicy> expiryPolicy;