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, 13 insertions, 7 deletions

diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
index cefea00262..048008f2a5 100644
--- a/cpp/src/qpid/cluster/Connection.h
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -64,10 +64,10 @@ class Connection :
   public:
     typedef sys::PollableQueue<EventFrame> PollableFrameQueue;
 
-    /** Local connection, use this in ConnectionId */
-    Connection(Cluster&, sys::ConnectionOutputHandler& out, const std::string& id, MemberId, bool catchUp, bool isLink);
-    /** Shadow connection */
-    Connection(Cluster&, sys::ConnectionOutputHandler& out, const std::string& id, ConnectionId);
+    /** Local connection. */
+    Connection(Cluster&, sys::ConnectionOutputHandler& out, const std::string& logId, MemberId, bool catchUp, bool isLink);
+    /** Shadow connection. */
+    Connection(Cluster&, sys::ConnectionOutputHandler& out, const std::string& logId, const ConnectionId& id);
     ~Connection();
     
     ConnectionId getId() const { return self; }
@@ -100,9 +100,12 @@ class Connection :
     /** Called if the connectors member has left the cluster */
     void left();
     
-    // ConnectionCodec methods
+    // ConnectionCodec methods - called by IO layer with a read buffer.
     size_t decode(const char* buffer, size_t size);
 
+    // Decode a data event, a read buffer that has been delivered by the cluster.
+    void decode(const EventHeader& eh, const void* data);
+
     // Called for data delivered from the cluster.
     void deliveredFrame(const EventFrame&);
 
@@ -118,7 +121,7 @@ class Connection :
                       const framing::SequenceNumber& received,
                       const framing::SequenceSet& unknownCompleted, const SequenceSet& receivedIncomplete);
     
-    void shadowReady(uint64_t memberId, uint64_t connectionId, const std::string& username);
+    void shadowReady(uint64_t memberId, uint64_t connectionId, const std::string& username, const std::string& fragment);
 
     void membership(const framing::FieldTable&, const framing::FieldTable&, uint64_t frameId);
 
@@ -149,7 +152,9 @@ class Connection :
     void exchange(const std::string& encoded);
 
     void giveReadCredit(int credit);
-    
+
+    framing::FrameDecoder& getDecoder() { return clusterDecoder; }
+
   private:
     struct NullFrameHandler : public framing::FrameHandler {
         void handle(framing::AMQFrame&) {}
@@ -174,6 +179,7 @@ class Connection :
     WriteEstimate writeEstimate;
     OutputInterceptor output;
     framing::FrameDecoder localDecoder;
+    framing::FrameDecoder clusterDecoder;
     broker::Connection connection;
     framing::SequenceNumber deliverSeq;
     framing::ChannelId currentChannel;